Add end-to-end Parquet VARIANT support

.github/workflows/scripts/.pinot_quickstart.sh

@@ -319,6 +319,41 @@ if [ "${PASS}" -eq 0 ]; then
   exit 1
 fi
 
+# Test the packaged Parquet VARIANT quickstart entrypoint
+bin/quick-start-variant-batch.sh &
+PID=$!
+
+PASS=0
+
+# Wait at most 5 minutes for the sample segment and a nested VARIANT value to be queryable
+for i in $(seq 1 150)
+do
+  QUERY_RES=$(curl -sS -X POST --header 'Accept: application/json' \
+    -d "{\"sql\":\"SET enableNullHandling=true; SELECT COUNT(*) FROM variantEvents\",\"trace\":false}" \
+    http://localhost:8000/query/sql)
+  if [ $? -eq 0 ]; then
+    COUNT_STAR_RES=$(echo "${QUERY_RES}" | jq '.resultTable.rows[0][0]')
+    VARIANT_QUERY_RES=$(curl -sS -X POST --header 'Accept: application/json' \
+      -d "{\"sql\":\"SET enableNullHandling=true; SELECT variant_get(payload, '$.user.id', 'STRING') FROM variantEvents WHERE eventId = 'evt-001'\",\"trace\":false}" \
+      http://localhost:8000/query/sql)
+    if [ $? -eq 0 ]; then
+      USER_ID_RES=$(echo "${VARIANT_QUERY_RES}" | jq -r '.resultTable.rows[0][0] // empty')
+      if [[ "${COUNT_STAR_RES}" =~ ^[0-9]+$ ]] && [ "${COUNT_STAR_RES}" -eq 5 ] \
+          && [ "${USER_ID_RES}" = "u-1" ]; then
+        PASS=1
+        break
+      fi
+    fi
+  fi
+  sleep 2
+done
+
+cleanup "${PID}"
+if [ "${PASS}" -eq 0 ]; then
+  echo 'Parquet VARIANT quickstart failed: Cannot query the five sample rows and nested user ID.'
+  exit 1
+fi
+
 # Test quick-start-streaming
 bin/quick-start-streaming.sh &
 PID=$!

compatibility-verifier/compCheck.sh

@@ -22,6 +22,8 @@
 # from one version to the other given 2 commit hashes. It first builds
 # Pinot in the 2 given directories and then upgrades in the following order:
 # Controller -> Broker -> Server
+# An optional old-broker-new-servers.yaml phase temporarily restores the old
+# broker after all servers are upgraded to verify the opposite wire direction.
 #
 # TODO Some ideas to explore:
 #  It will be nice to have the script take arguments about what is to be done.
@@ -496,14 +498,24 @@ setupControllerVariables
 setupBrokerVariables
 setupServerVariables
 
-export JAVA_OPTS="-DControllerPort=${CONTROLLER_PORT} -DBrokerQueryPort=${BROKER_QUERY_PORT} -DServerAdminPort=${SERVER_ADMIN_PORT}"
-
 mkdir ${PID_DIR}
 mkdir ${LOG_DIR}
 
 oldTargetDir="$workingDir"/oldTargetDir
 newTargetDir="$workingDir"/newTargetDir
 
+oldServerSupportsVariant=false
+oldExpressionsProto="${oldTargetDir}/pinot-common/src/main/proto/expressions.proto"
+if [ -f "${oldExpressionsProto}" ] \
+    && grep -Eq '^[[:space:]]*VARIANT[[:space:]]*=[[:space:]]*24[[:space:]]*;' "${oldExpressionsProto}"; then
+  oldServerSupportsVariant=true
+fi
+echo "Old server supports VARIANT query-wire type: ${oldServerSupportsVariant}"
+
+export JAVA_OPTS="-DControllerPort=${CONTROLLER_PORT} -DBrokerQueryPort=${BROKER_QUERY_PORT} \
+-DServerAdminPort=${SERVER_ADMIN_PORT} -Dpinot.compat.oldServerSupportsVariant=${oldServerSupportsVariant} \
+-Dpinot.compat.logDir=${LOG_DIR}"
+
 setupCompatTester
 
 # check that the default ports are open
@@ -604,7 +616,30 @@ if [ -f "${SERVER_CONF_2}" ]; then
       exit 1
     fi
   fi
+fi
+
+if [ -f "$testSuiteDir/old-broker-new-servers.yaml" ]; then
+  echo "Temporarily downgrading broker to test the old broker with upgraded servers"
+  stopService broker
+  startService broker "$oldTargetDir" "$BROKER_CONF"
+  waitForBrokerReady
+
+  "$COMPAT_TESTER" "$testSuiteDir/old-broker-new-servers.yaml" "$genNum"
+  if [ $? -ne 0 ]; then
+    if [ $keepClusterOnFailure == "false" ]; then
+      stopServices
+    fi
+    echo "Failed with old broker and upgraded servers"
+    exit 1
+  fi
 
+  echo "Restoring upgraded broker before server rollback"
+  stopService broker
+  startService broker "$newTargetDir" "$BROKER_CONF"
+  waitForBrokerReady
+fi
+
+if [ -f "${SERVER_CONF_2}" ]; then
   echo "Downgrading server 2"
   # Upgrade completed, now do a rollback
   stopService server2

compatibility-verifier/multi-stage-query-engine-test-suite/config/queries/routing-ready.queries

@@ -0,0 +1,21 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+SELECT COUNT(*) FROM FeatureTest1 WHERE generationNumber = __GENERATION_NUMBER__
+SELECT COUNT(*) FROM FeatureTest2 WHERE generationNumber = __GENERATION_NUMBER__

compatibility-verifier/multi-stage-query-engine-test-suite/config/queries/variant-wire-mixed.queries

@@ -0,0 +1,21 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+# Scan all four uploaded segments so the leaf stage reaches both the upgraded and old server.
+SELECT COUNT(*) FROM FeatureTest1 WHERE variant_get(parse_json(concat(concat('"', stringDimSV1), '"')), '$', 'STRING') = stringDimSV1

compatibility-verifier/multi-stage-query-engine-test-suite/config/queries/variant-wire.queries

@@ -0,0 +1,21 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+# Keep the VARIANT-producing call in a leaf filter so the server is part of the wire-compatibility check.
+SELECT COUNT(*) FROM FeatureTest1 WHERE generationNumber = __GENERATION_NUMBER__ AND variant_get(parse_json(concat(concat('"', stringDimSV1), '"')), '$', 'STRING') = stringDimSV1

compatibility-verifier/multi-stage-query-engine-test-suite/config/query-results/routing-ready.results

@@ -0,0 +1,21 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+{"resultTable":{"dataSchema":{"columnNames":["EXPR$0"],"columnDataTypes":["LONG"]},"rows":[[10]]},"exceptions":[],"numDocsScanned":10}
+{"resultTable":{"dataSchema":{"columnNames":["EXPR$0"],"columnDataTypes":["LONG"]},"rows":[[66]]},"exceptions":[],"numDocsScanned":66}

compatibility-verifier/multi-stage-query-engine-test-suite/config/query-results/variant-wire-mixed.results

@@ -0,0 +1,20 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+{"resultTable":{"dataSchema":{"columnNames":["EXPR$0"],"columnDataTypes":["LONG"]},"rows":[[40]]},"exceptions":[],"numDocsScanned":40}

compatibility-verifier/multi-stage-query-engine-test-suite/config/query-results/variant-wire.results

@@ -0,0 +1,20 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+{"resultTable":{"dataSchema":{"columnNames":["EXPR$0"],"columnDataTypes":["LONG"]},"rows":[[10]]},"exceptions":[],"numDocsScanned":10}

compatibility-verifier/multi-stage-query-engine-test-suite/old-broker-new-servers.yaml

@@ -0,0 +1,33 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+description: Operations to be run with the old broker and upgraded servers
+operations:
+  - type: queryOp
+    description: Wait for the restarted old broker to rebuild its routing tables
+    useMultiStageQueryEngine: true
+    queryFileName: queries/routing-ready.queries
+    expectedResultsFileName: query-results/routing-ready.results
+    maxAttempts: 60
+    retryDelayMs: 1000
+  - type: queryOp
+    description: Verify the old broker's pre-VARIANT plans remain compatible with upgraded servers
+    useMultiStageQueryEngine: true
+    queryFileName: queries/feature-test-multi-stage.queries
+    expectedResultsFileName: query-results/feature-test-multi-stage.results

compatibility-verifier/multi-stage-query-engine-test-suite/post-server-2-upgrade.yaml

@@ -41,3 +41,9 @@ operations:
     useMultiStageQueryEngine: true
     queryFileName: queries/feature-test-multi-stage.queries
     expectedResultsFileName: query-results/feature-test-multi-stage.results
+  - type: queryOp
+    description: Verify the VARIANT wire query after both servers are upgraded
+    useMultiStageQueryEngine: true
+    nullHandlingEnabled: true
+    queryFileName: queries/variant-wire.queries
+    expectedResultsFileName: query-results/variant-wire.results

compatibility-verifier/multi-stage-query-engine-test-suite/post-server-upgrade.yaml

@@ -41,3 +41,19 @@ operations:
     useMultiStageQueryEngine: true
     queryFileName: queries/feature-test-multi-stage.queries
     expectedResultsFileName: query-results/feature-test-multi-stage.results
+  - type: queryOp
+    description: Reject the VARIANT wire query without partial rows while the server fleet is mixed
+    runIfSystemProperty: pinot.compat.oldServerSupportsVariant
+    runIfSystemPropertyValue: "false"
+    useMultiStageQueryEngine: true
+    nullHandlingEnabled: true
+    queryFileName: queries/variant-wire-mixed.queries
+    expectedErrorMessageContains: "Caught exception while deserializing stage plan"
+  - type: queryOp
+    description: Run the VARIANT wire query across the mixed fleet when the old server supports the type
+    runIfSystemProperty: pinot.compat.oldServerSupportsVariant
+    runIfSystemPropertyValue: "true"
+    useMultiStageQueryEngine: true
+    nullHandlingEnabled: true
+    queryFileName: queries/variant-wire-mixed.queries
+    expectedResultsFileName: query-results/variant-wire-mixed.results

compatibility-verifier/multi-stage-query-engine-test-suite/pre-server-upgrade.yaml

@@ -41,3 +41,26 @@ operations:
     useMultiStageQueryEngine: true
     queryFileName: queries/feature-test-multi-stage.queries
     expectedResultsFileName: query-results/feature-test-multi-stage.results
+  - type: queryOp
+    description: Report the deterministic VARIANT wire-type error while servers are still on the old version
+    runIfSystemProperty: pinot.compat.oldServerSupportsVariant
+    runIfSystemPropertyValue: "false"
+    useMultiStageQueryEngine: true
+    nullHandlingEnabled: true
+    queryFileName: queries/variant-wire.queries
+    expectedErrorMessageContains: "Caught exception while deserializing stage plan"
+  - type: queryOp
+    description: Run the VARIANT wire query when the old servers already advertise the type
+    runIfSystemProperty: pinot.compat.oldServerSupportsVariant
+    runIfSystemPropertyValue: "true"
+    useMultiStageQueryEngine: true
+    nullHandlingEnabled: true
+    queryFileName: queries/variant-wire.queries
+    expectedResultsFileName: query-results/variant-wire.results
+  - type: fileContainsOp
+    description: Verify the exact unsupported VARIANT wire type in the old server log
+    runIfSystemProperty: pinot.compat.oldServerSupportsVariant
+    runIfSystemPropertyValue: "false"
+    directorySystemProperty: pinot.compat.logDir
+    fileNameGlob: "server*.log"
+    expectedTextContains: "Unsupported proto ColumnDataType: UNRECOGNIZED"

compatibility-verifier/sample-test-suite/config/queries/routing-ready.queries

@@ -0,0 +1,20 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+SELECT COUNT(*) FROM FeatureTest1 WHERE generationNumber = __GENERATION_NUMBER__

compatibility-verifier/sample-test-suite/config/query-results/routing-ready.results

@@ -0,0 +1,20 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+{"resultTable":{"dataSchema":{"columnDataTypes":["LONG"],"columnNames":["count(*)"]},"rows":[[10]]},"exceptions":[],"numServersQueried":1,"numServersResponded":1,"numSegmentsQueried":1,"numSegmentsProcessed":1,"numSegmentsMatched":1,"numDocsScanned":10,"numEntriesScannedPostFilter":0,"numGroupsLimitReached":false,"totalDocs":10,"timeUsedMs":4,"segmentStatistics":[],"traceInfo":{},"minConsumingFreshnessTimeMs":0}