This is an automated email from the ASF dual-hosted git repository. xiangfu0 pushed a commit to branch xiangfu0/variant-split/08-compat-e2e in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 47c2a545bae71d543e19bf803b87f7ddc96e9c16 Author: Xiang Fu <[email protected]> AuthorDate: Sat Oct 24 16:22:31 2026 -0700 [VARIANT 8/8] VARIANT compatibility suites, integration test and quickstart Adds mixed-version compatibility suites for the VARIANT wire type, an end-to-end integration test, and a Parquet VARIANT batch quickstart. --- .github/workflows/scripts/.pinot_quickstart.sh | 35 ++ compatibility-verifier/compCheck.sh | 50 +- .../config/queries/routing-ready.queries | 21 + .../config/queries/variant-wire-mixed.queries | 21 + .../config/queries/variant-wire.queries | 21 + .../config/query-results/routing-ready.results | 21 + .../query-results/variant-wire-mixed.results | 20 + .../config/query-results/variant-wire.results | 20 + ...-2-upgrade.yaml => old-broker-new-servers.yaml} | 28 +- .../post-server-2-upgrade.yaml | 6 + .../post-server-upgrade.yaml | 16 + .../pre-server-upgrade.yaml | 23 + .../config/queries/routing-ready.queries | 20 + .../config/query-results/routing-ready.results | 20 + .../sample-test-suite/old-broker-new-servers.yaml | 31 + .../main/java/org/apache/pinot/compat/BaseOp.java | 48 +- .../pinot/compat/CompatibilityOpsRunner.java | 3 + .../org/apache/pinot/compat/FileContainsOp.java | 136 +++++ .../main/java/org/apache/pinot/compat/QueryOp.java | 205 ++++++- .../main/java/org/apache/pinot/compat/Utils.java | 24 +- .../java/org/apache/pinot/compat/BaseOpTest.java | 78 +++ .../apache/pinot/compat/FileContainsOpTest.java | 57 ++ .../java/org/apache/pinot/compat/QueryOpTest.java | 196 ++++++ .../pinot/integration/tests/ClusterTest.java | 1 + .../integration/tests/custom/VariantTypeTest.java | 661 +++++++++++++++++++++ pinot-tools/pom.xml | 10 + .../org/apache/pinot/tools/VariantQuickStart.java | 114 ++++ .../examples/batch/variantEvents/README.md | 88 +++ .../batch/variantEvents/ingestionJobSpec.yaml | 51 ++ .../rawdata/variantEvents_data.parquet | Bin 0 -> 1442 bytes .../variantEvents_offline_table_config.json | 39 ++ .../batch/variantEvents/variantEvents_schema.json | 26 + .../pinot/tools/admin/command/QuickStartTest.java | 7 + 33 files changed, 2045 insertions(+), 52 deletions(-) diff --git a/.github/workflows/scripts/.pinot_quickstart.sh b/.github/workflows/scripts/.pinot_quickstart.sh index dad7e5fb258..82b4954d3fd 100755 --- a/.github/workflows/scripts/.pinot_quickstart.sh +++ b/.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=$! diff --git a/compatibility-verifier/compCheck.sh b/compatibility-verifier/compCheck.sh index 7ab06270fb8..3f37212f2ab 100755 --- a/compatibility-verifier/compCheck.sh +++ b/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,33 @@ 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 +oldCommonJar=$(find "${oldTargetDir}/pinot-common/target" -maxdepth 1 -type f -name 'pinot-common-*.jar' \ + ! -name '*-sources.jar' ! -name '*-tests.jar' ! -name '*-javadoc.jar' -print -quit) +if [ -z "${oldCommonJar}" ]; then + echo "Failed to locate the built old pinot-common artifact under ${oldTargetDir}/pinot-common/target" + exit 1 +fi +if ! oldColumnDataType=$(javap -classpath "${oldCommonJar}" \ + 'org.apache.pinot.common.proto.Expressions$ColumnDataType' 2>&1); then + echo "Failed to inspect ColumnDataType in built old artifact ${oldCommonJar}: ${oldColumnDataType}" + exit 1 +fi +if echo "${oldColumnDataType}" | grep -Eq 'public static final .*ColumnDataType VARIANT;'; 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 +625,32 @@ 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 + + # This phase contains only queries against the data generated by the preceding upgrade phase, so intentionally + # reuse genNum instead of generating another table/segment set while the old broker is active. + "$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 diff --git a/compatibility-verifier/multi-stage-query-engine-test-suite/config/queries/routing-ready.queries b/compatibility-verifier/multi-stage-query-engine-test-suite/config/queries/routing-ready.queries new file mode 100644 index 00000000000..88a2c45b180 --- /dev/null +++ b/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__ diff --git a/compatibility-verifier/multi-stage-query-engine-test-suite/config/queries/variant-wire-mixed.queries b/compatibility-verifier/multi-stage-query-engine-test-suite/config/queries/variant-wire-mixed.queries new file mode 100644 index 00000000000..81f0b914dcb --- /dev/null +++ b/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 diff --git a/compatibility-verifier/multi-stage-query-engine-test-suite/config/queries/variant-wire.queries b/compatibility-verifier/multi-stage-query-engine-test-suite/config/queries/variant-wire.queries new file mode 100644 index 00000000000..81d019ae145 --- /dev/null +++ b/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 diff --git a/compatibility-verifier/multi-stage-query-engine-test-suite/config/query-results/routing-ready.results b/compatibility-verifier/multi-stage-query-engine-test-suite/config/query-results/routing-ready.results new file mode 100644 index 00000000000..53d8cdb48f0 --- /dev/null +++ b/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} diff --git a/compatibility-verifier/multi-stage-query-engine-test-suite/config/query-results/variant-wire-mixed.results b/compatibility-verifier/multi-stage-query-engine-test-suite/config/query-results/variant-wire-mixed.results new file mode 100644 index 00000000000..8f6183fa0d5 --- /dev/null +++ b/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} diff --git a/compatibility-verifier/multi-stage-query-engine-test-suite/config/query-results/variant-wire.results b/compatibility-verifier/multi-stage-query-engine-test-suite/config/query-results/variant-wire.results new file mode 100644 index 00000000000..d573b9e73d9 --- /dev/null +++ b/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} diff --git a/compatibility-verifier/multi-stage-query-engine-test-suite/post-server-2-upgrade.yaml b/compatibility-verifier/multi-stage-query-engine-test-suite/old-broker-new-servers.yaml similarity index 53% copy from compatibility-verifier/multi-stage-query-engine-test-suite/post-server-2-upgrade.yaml copy to compatibility-verifier/multi-stage-query-engine-test-suite/old-broker-new-servers.yaml index 6dbb916845c..5924c42c15e 100644 --- a/compatibility-verifier/multi-stage-query-engine-test-suite/post-server-2-upgrade.yaml +++ b/compatibility-verifier/multi-stage-query-engine-test-suite/old-broker-new-servers.yaml @@ -17,27 +17,17 @@ # under the License. # -# Operations to be done. -description: Operations to be run after server upgrade +description: Operations to be run with the old broker and upgraded servers operations: - - type: segmentOp - description: Add segment FeatureTest1_Segment5 to table FeatureTest1 - op: UPLOAD - inputDataFileName: data/FeatureTest1-data-00.csv - schemaFileName: FeatureTest1-schema.json - tableConfigFileName: feature-test-1.json - recordReaderConfigFileName: data/recordReaderConfig.json - segmentName: FeatureTest1_Segment5 - - type: streamOp - description: publish rows to PinotRealtimeFeatureTest2Event - op: PRODUCE - streamConfigFileName: feature-test-2-realtime-stream-config.json - numRows: 66 - inputDataFileName: data/FeatureTest2-data-realtime-00.csv - recordReaderConfigFileName: data/recordReaderConfig.json - tableConfigFileName: feature-test-2-realtime.json - type: queryOp - description: Run multi-stage queries on FeatureTest1 and FeatureTest2 using SQL + 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 diff --git a/compatibility-verifier/multi-stage-query-engine-test-suite/post-server-2-upgrade.yaml b/compatibility-verifier/multi-stage-query-engine-test-suite/post-server-2-upgrade.yaml index 6dbb916845c..ef0aa500058 100644 --- a/compatibility-verifier/multi-stage-query-engine-test-suite/post-server-2-upgrade.yaml +++ b/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 diff --git a/compatibility-verifier/multi-stage-query-engine-test-suite/post-server-upgrade.yaml b/compatibility-verifier/multi-stage-query-engine-test-suite/post-server-upgrade.yaml index 786cb3fb398..86b02dfc13a 100644 --- a/compatibility-verifier/multi-stage-query-engine-test-suite/post-server-upgrade.yaml +++ b/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 diff --git a/compatibility-verifier/multi-stage-query-engine-test-suite/pre-server-upgrade.yaml b/compatibility-verifier/multi-stage-query-engine-test-suite/pre-server-upgrade.yaml index 167f71fadf2..2b215703390 100644 --- a/compatibility-verifier/multi-stage-query-engine-test-suite/pre-server-upgrade.yaml +++ b/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" diff --git a/compatibility-verifier/sample-test-suite/config/queries/routing-ready.queries b/compatibility-verifier/sample-test-suite/config/queries/routing-ready.queries new file mode 100644 index 00000000000..7bac823c16a --- /dev/null +++ b/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__ diff --git a/compatibility-verifier/sample-test-suite/config/query-results/routing-ready.results b/compatibility-verifier/sample-test-suite/config/query-results/routing-ready.results new file mode 100644 index 00000000000..60667c7ec5f --- /dev/null +++ b/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} diff --git a/compatibility-verifier/sample-test-suite/old-broker-new-servers.yaml b/compatibility-verifier/sample-test-suite/old-broker-new-servers.yaml new file mode 100644 index 00000000000..5b6819ee341 --- /dev/null +++ b/compatibility-verifier/sample-test-suite/old-broker-new-servers.yaml @@ -0,0 +1,31 @@ +# +# 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 the FeatureTest1 routing table + queryFileName: queries/routing-ready.queries + expectedResultsFileName: query-results/routing-ready.results + maxAttempts: 60 + retryDelayMs: 1000 + - type: queryOp + description: Verify single-stage queries through the old broker and upgraded servers + queryFileName: queries/feature-test-1-sql.queries + expectedResultsFileName: query-results/feature-test-1-rest-sql.results diff --git a/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/BaseOp.java b/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/BaseOp.java index 2869adda0dc..061f5d58809 100644 --- a/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/BaseOp.java +++ b/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/BaseOp.java @@ -22,6 +22,7 @@ import com.fasterxml.jackson.annotation.JsonSubTypes; import com.fasterxml.jackson.annotation.JsonTypeInfo; import java.net.URI; import java.util.Properties; +import javax.annotation.Nullable; import org.apache.pinot.client.PinotClientException; import org.apache.pinot.client.admin.PinotAdminClient; import org.apache.pinot.client.admin.PinotAdminTransport; @@ -34,11 +35,12 @@ import org.slf4j.LoggerFactory; @JsonSubTypes.Type(value = SegmentOp.class, name = "segmentOp"), @JsonSubTypes.Type(value = TableOp.class, name = "tableOp"), @JsonSubTypes.Type(value = QueryOp.class, name = "queryOp"), - @JsonSubTypes.Type(value = StreamOp.class, name = "streamOp") + @JsonSubTypes.Type(value = StreamOp.class, name = "streamOp"), + @JsonSubTypes.Type(value = FileContainsOp.class, name = "fileContainsOp") }) public abstract class BaseOp { enum OpType { - TABLE_OP, SEGMENT_OP, QUERY_OP, STREAM_OP, + TABLE_OP, SEGMENT_OP, QUERY_OP, STREAM_OP, FILE_CONTAINS_OP, } private String _name; @@ -48,6 +50,10 @@ public abstract class BaseOp { protected static final String GENERATION_NUMBER_PLACEHOLDER = "__GENERATION_NUMBER__"; protected static final String CONFIG_PLACEHOLDER = "/config/"; private String _parentDir; + @Nullable + private String _runIfSystemProperty; + @Nullable + private String _runIfSystemPropertyValue; protected BaseOp(OpType opType) { _opType = opType; @@ -81,7 +87,45 @@ public abstract class BaseOp { return _parentDir + CONFIG_PLACEHOLDER + fileName; } + /// Returns the system property that gates this operation, or {@code null} when the operation is unconditional. + @Nullable + public String getRunIfSystemProperty() { + return _runIfSystemProperty; + } + + public void setRunIfSystemProperty(@Nullable String runIfSystemProperty) { + _runIfSystemProperty = runIfSystemProperty; + } + + /// Returns the exact property value required to run this operation, or {@code null} when the operation is + /// unconditional. + @Nullable + public String getRunIfSystemPropertyValue() { + return _runIfSystemPropertyValue; + } + + public void setRunIfSystemPropertyValue(@Nullable String runIfSystemPropertyValue) { + _runIfSystemPropertyValue = runIfSystemPropertyValue; + } + public boolean run(int generationNumber) { + if ((_runIfSystemProperty == null) != (_runIfSystemPropertyValue == null) + || _runIfSystemProperty != null && (_runIfSystemProperty.isBlank() || _runIfSystemPropertyValue.isBlank())) { + LOGGER.error("Both runIfSystemProperty and a non-blank runIfSystemPropertyValue must be configured together"); + return false; + } + if (_runIfSystemProperty != null) { + String actualValue = System.getProperty(_runIfSystemProperty); + if (actualValue == null) { + LOGGER.error("Required system property {} is not configured", _runIfSystemProperty); + return false; + } + if (!_runIfSystemPropertyValue.equals(actualValue)) { + LOGGER.info("Skipping OpType {} because system property {} is '{}', expected '{}'", _opType, + _runIfSystemProperty, actualValue, _runIfSystemPropertyValue); + return true; + } + } LOGGER.info("Running OpType {} : {}", _opType.toString(), getDescription()); return runOp(generationNumber); } diff --git a/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/CompatibilityOpsRunner.java b/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/CompatibilityOpsRunner.java index eba60c62c0f..6d6c2af132c 100644 --- a/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/CompatibilityOpsRunner.java +++ b/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/CompatibilityOpsRunner.java @@ -115,6 +115,9 @@ public class CompatibilityOpsRunner { case "streamOp": node.setType(StreamOp.class); break; + case "fileContainsOp": + node.setType(FileContainsOp.class); + break; default: throw new RuntimeException("Unknown type: " + type); } diff --git a/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/FileContainsOp.java b/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/FileContainsOp.java new file mode 100644 index 00000000000..343228b7d7f --- /dev/null +++ b/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/FileContainsOp.java @@ -0,0 +1,136 @@ +/** + * 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. + */ +package org.apache.pinot.compat; + +import java.io.BufferedReader; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.DirectoryStream; +import java.nio.file.Files; +import java.nio.file.Path; +import javax.annotation.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/// Verifies that at least one file matching a glob in a system-property directory contains an expected text fragment. +/// Instances are configured and invoked serially by the compatibility runner and are not thread-safe. +public class FileContainsOp extends BaseOp { + private static final Logger LOGGER = LoggerFactory.getLogger(FileContainsOp.class); + private static final int MAX_ATTEMPTS = 10; + private static final long RETRY_DELAY_MS = 100L; + + @Nullable + private String _directorySystemProperty; + @Nullable + private String _fileNameGlob; + @Nullable + private String _expectedTextContains; + + public FileContainsOp() { + super(OpType.FILE_CONTAINS_OP); + } + + @Nullable + public String getDirectorySystemProperty() { + return _directorySystemProperty; + } + + public void setDirectorySystemProperty(@Nullable String directorySystemProperty) { + _directorySystemProperty = directorySystemProperty; + } + + @Nullable + public String getFileNameGlob() { + return _fileNameGlob; + } + + public void setFileNameGlob(@Nullable String fileNameGlob) { + _fileNameGlob = fileNameGlob; + } + + @Nullable + public String getExpectedTextContains() { + return _expectedTextContains; + } + + public void setExpectedTextContains(@Nullable String expectedTextContains) { + _expectedTextContains = expectedTextContains; + } + + @Override + boolean runOp(int generationNumber) { + if (isBlank(_directorySystemProperty) || isBlank(_fileNameGlob) || isBlank(_expectedTextContains)) { + LOGGER.error( + "directorySystemProperty, fileNameGlob, and expectedTextContains must all be configured and non-blank"); + return false; + } + String directoryName = System.getProperty(_directorySystemProperty); + if (isBlank(directoryName)) { + LOGGER.error("Required log-directory system property {} is not configured", _directorySystemProperty); + return false; + } + Path directory = Path.of(directoryName); + if (!Files.isDirectory(directory)) { + LOGGER.error("Configured log directory does not exist: {}", directory); + return false; + } + + try { + for (int attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) { + if (containsExpectedText(directory, _fileNameGlob, _expectedTextContains)) { + return true; + } + if (attempt < MAX_ATTEMPTS) { + Thread.sleep(RETRY_DELAY_MS); + } + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.error("Interrupted while waiting for expected text in {} files under {}", _fileNameGlob, directory); + return false; + } catch (IOException e) { + LOGGER.error("Failed to inspect {} files under {}", _fileNameGlob, directory, e); + return false; + } + LOGGER.error("No {} file under {} contained expected text: {}", _fileNameGlob, directory, _expectedTextContains); + return false; + } + + static boolean containsExpectedText(Path directory, String fileNameGlob, String expectedText) + throws IOException { + try (DirectoryStream<Path> paths = Files.newDirectoryStream(directory, fileNameGlob)) { + for (Path path : paths) { + try (BufferedReader reader = Files.newBufferedReader(path, StandardCharsets.UTF_8)) { + String line; + while ((line = reader.readLine()) != null) { + if (line.contains(expectedText)) { + return true; + } + } + } + } + } + return false; + } + + private static boolean isBlank(String value) { + return value == null || value.isBlank(); + } +} diff --git a/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/QueryOp.java b/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/QueryOp.java index 536ab1024e7..101bc2dedef 100644 --- a/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/QueryOp.java +++ b/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/QueryOp.java @@ -22,8 +22,10 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties; import com.fasterxml.jackson.databind.JsonNode; import java.io.BufferedReader; import java.io.FileInputStream; +import java.io.IOException; import java.io.InputStreamReader; import java.nio.charset.StandardCharsets; +import javax.annotation.Nullable; import org.apache.pinot.common.utils.SqlResultComparator; import org.apache.pinot.spi.utils.JsonUtils; import org.apache.pinot.tools.utils.ExplainPlanUtils; @@ -31,8 +33,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -/// Executes queries in the query file, and compares the results with the ones given in the -/// expected results file +/// Executes queries in the query file and verifies either result rows or an expected error-message substring. +/// Exactly one of `expectedResultsFileName` and `expectedErrorMessageContains` must be configured. /// /// TODO: /// - If we use current timestamp for realtime tables, we may not be able to use pre-canned queries. @@ -42,8 +44,14 @@ public class QueryOp extends BaseOp { private static final String COMMENT_DELIMITER = "#"; private String _queryFileName; + @Nullable private String _expectedResultsFileName; + @Nullable + private String _expectedErrorMessageContains; private boolean _useMultiStageQueryEngine = false; + private boolean _enableNullHandling = false; + private int _maxAttempts = 1; + private long _retryDelayMs = 1000L; public QueryOp() { super(OpType.QUERY_OP); @@ -62,14 +70,28 @@ public class QueryOp extends BaseOp { _queryFileName = queryFileName; } + /// Returns the expected-results file, or {@code null} when expected-error mode is configured. + @Nullable public String getExpectedResultsFileName() { return _expectedResultsFileName; } - public void setExpectedResultsFileName(String expectedResultsFileName) { + /// Configures result-comparison mode. Pass {@code null} to clear it before configuring expected-error mode. + public void setExpectedResultsFileName(@Nullable String expectedResultsFileName) { _expectedResultsFileName = expectedResultsFileName; } + /// Returns the required error-message substring, or {@code null} when result-comparison mode is configured. + @Nullable + public String getExpectedErrorMessageContains() { + return _expectedErrorMessageContains; + } + + /// Configures expected-error mode. Pass {@code null} to clear it before configuring result-comparison mode. + public void setExpectedErrorMessageContains(@Nullable String expectedErrorMessageContains) { + _expectedErrorMessageContains = expectedErrorMessageContains; + } + public boolean getUseMultiStageQueryEngine() { return _useMultiStageQueryEngine; } @@ -78,31 +100,99 @@ public class QueryOp extends BaseOp { _useMultiStageQueryEngine = useMultiStageQueryEngine; } + public boolean isNullHandlingEnabled() { + return _enableNullHandling; + } + + public void setNullHandlingEnabled(boolean nullHandlingEnabled) { + _enableNullHandling = nullHandlingEnabled; + } + + public int getMaxAttempts() { + return _maxAttempts; + } + + public void setMaxAttempts(int maxAttempts) { + _maxAttempts = maxAttempts; + } + + public long getRetryDelayMs() { + return _retryDelayMs; + } + + public void setRetryDelayMs(long retryDelayMs) { + _retryDelayMs = retryDelayMs; + } + @Override boolean runOp(int generationNumber) { - LOGGER.info("Verifying queries in {} against results in {}", _queryFileName, _expectedResultsFileName); + if (!hasValidExpectedOutcomeConfiguration()) { + LOGGER.error( + "Exactly one of expectedResultsFileName or expectedErrorMessageContains must be configured for queries in {}", + _queryFileName); + return false; + } + if (_maxAttempts < 1 || _retryDelayMs < 0) { + LOGGER.error("maxAttempts must be positive and retryDelayMs must be non-negative for queries in {}", + _queryFileName); + return false; + } + if (_expectedErrorMessageContains != null) { + LOGGER.info("Verifying queries in {} fail with an error containing '{}'", _queryFileName, + _expectedErrorMessageContains); + } else { + LOGGER.info("Verifying queries in {} against results in {}", _queryFileName, _expectedResultsFileName); + } try { for (int i = 1; i <= generationNumber; i++) { - if (!verifyQueries(i)) { + if (!verifyQueriesWithRetry(i)) { return false; } } return true; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.error("Interrupted while retrying queries in {}", _queryFileName); + return false; } catch (Exception e) { LOGGER.error("FAILED to verify queries in {}: {}", _queryFileName, e); return false; } } + private boolean verifyQueriesWithRetry(int generationNumber) + throws Exception { + for (int attempt = 1; attempt <= _maxAttempts; attempt++) { + if (verifyQueries(generationNumber)) { + return true; + } + if (attempt < _maxAttempts) { + LOGGER.info("Retrying queries in {} after failed attempt {} of {}", _queryFileName, attempt, _maxAttempts); + Thread.sleep(_retryDelayMs); + } + } + return false; + } + + boolean hasValidExpectedOutcomeConfiguration() { + if (_expectedResultsFileName != null && _expectedErrorMessageContains != null) { + return false; + } + boolean hasExpectedResults = _expectedResultsFileName != null && !_expectedResultsFileName.isBlank(); + boolean hasExpectedError = + _expectedErrorMessageContains != null && !_expectedErrorMessageContains.isBlank(); + return hasExpectedResults != hasExpectedError; + } + boolean verifyQueries(int generationNumber) throws Exception { boolean testPassed = false; try (BufferedReader queryReader = new BufferedReader( new InputStreamReader(new FileInputStream(getAbsoluteFileName(_queryFileName)), StandardCharsets.UTF_8)); - BufferedReader expectedResultReader = new BufferedReader( - new InputStreamReader(new FileInputStream(getAbsoluteFileName(_expectedResultsFileName)), - StandardCharsets.UTF_8))) { + BufferedReader expectedResultReader = _expectedErrorMessageContains != null ? null + : new BufferedReader(new InputStreamReader( + new FileInputStream(getAbsoluteFileName(_expectedResultsFileName)), StandardCharsets.UTF_8))) { int succeededQueryCount = 0; int totalQueryCount = 0; @@ -116,38 +206,47 @@ public class QueryOp extends BaseOp { } query = query.replaceAll(GENERATION_NUMBER_PLACEHOLDER, String.valueOf(generationNumber)); JsonNode expectedJson = null; - try { - String expectedResultLine = expectedResultReader.readLine(); - while (shouldIgnore(expectedResultLine)) { - expectedResultLine = expectedResultReader.readLine(); + if (_expectedErrorMessageContains == null) { + try { + String expectedResultLine = readNextExpectedResult(expectedResultReader, queryLineNum); + expectedJson = JsonUtils.stringToJsonNode(expectedResultLine); + } catch (Exception e) { + LOGGER.error("Comparison FAILED: Line: {} Exception caught while getting expected response for query: '{}'", + queryLineNum, query, e); } - expectedJson = JsonUtils.stringToJsonNode(expectedResultLine); - } catch (Exception e) { - LOGGER.error("Comparison FAILED: Line: {} Exception caught while getting expected response for query: '{}'", - queryLineNum, query, e); } JsonNode actualJson = null; - if (expectedJson != null) { + if (_expectedErrorMessageContains != null || expectedJson != null) { try { actualJson = _useMultiStageQueryEngine - ? Utils.postMultiStageSqlQuery(query, ClusterDescriptor.getInstance().getBrokerUrl()) - : Utils.postSqlQuery(query, ClusterDescriptor.getInstance().getBrokerUrl()); + ? Utils.postMultiStageSqlQuery(query, ClusterDescriptor.getInstance().getBrokerUrl(), + _enableNullHandling) + : Utils.postSqlQuery(query, ClusterDescriptor.getInstance().getBrokerUrl(), _enableNullHandling); } catch (Exception e) { LOGGER.error("Comparison FAILED: Line: {} Exception caught while running query: '{}', explain plan: {}", queryLineNum, query, getExplainPlan(query), e); } } - if (expectedJson != null && actualJson != null) { + if (actualJson != null && (_expectedErrorMessageContains != null || expectedJson != null)) { try { - boolean passed = _useMultiStageQueryEngine - ? SqlResultComparator.areMultiStageQueriesEqual(actualJson, expectedJson, query) - : SqlResultComparator.areEqual(actualJson, expectedJson, query); + boolean passed = matchesExpectedResponse(actualJson, expectedJson, _expectedErrorMessageContains, + _useMultiStageQueryEngine, query); if (passed) { succeededQueryCount++; - LOGGER.debug("Comparison PASSED: Line: {}, query: '{}', actual response: {}, expected response: {}", - queryLineNum, query, actualJson, expectedJson); + if (_expectedErrorMessageContains != null) { + LOGGER.debug("Comparison PASSED: Line: {}, query: '{}', actual response contains expected error: '{}'", + queryLineNum, query, _expectedErrorMessageContains); + } else { + LOGGER.debug("Comparison PASSED: Line: {}, query: '{}', actual response: {}, expected response: {}", + queryLineNum, query, actualJson, expectedJson); + } + } else if (_expectedErrorMessageContains != null) { + LOGGER.error( + "Comparison FAILED: Line: {}, query: '{}', actual response: {}, expected an exception containing: " + + "'{}'", + queryLineNum, query, actualJson, _expectedErrorMessageContains); } else { LOGGER.error( "Comparison FAILED: Line: {}, query: '{}', actual response: {}, expected response: {}, explain " @@ -172,16 +271,68 @@ public class QueryOp extends BaseOp { return testPassed; } + static String readNextExpectedResult(BufferedReader expectedResultReader, int queryLineNum) + throws IOException { + String expectedResultLine; + while ((expectedResultLine = expectedResultReader.readLine()) != null) { + if (!expectedResultLine.trim().isEmpty() && !expectedResultLine.trim().startsWith(COMMENT_DELIMITER)) { + return expectedResultLine; + } + } + throw new IOException("Expected results file ended before query at line " + queryLineNum); + } + + static boolean matchesExpectedResponse(JsonNode actual, @Nullable JsonNode expected, + @Nullable String expectedErrorMessageContains, boolean useMultiStageQueryEngine, String query) + throws IOException { + if (expectedErrorMessageContains != null) { + return hasExpectedError(actual, expectedErrorMessageContains); + } + return useMultiStageQueryEngine + ? SqlResultComparator.areMultiStageQueriesEqual(actual, expected, query) + : SqlResultComparator.areEqual(actual, expected, query); + } + + /// Returns whether a response has no result rows and contains only errors matching the required substring. + static boolean hasExpectedError(@Nullable JsonNode response, @Nullable String expectedErrorMessageContains) { + if (response == null || expectedErrorMessageContains == null || expectedErrorMessageContains.isEmpty()) { + return false; + } + JsonNode resultRows = response.path("resultTable").path("rows"); + if ((resultRows.isArray() && !resultRows.isEmpty()) || response.path("numRowsResultSet").asInt() > 0) { + return false; + } + JsonNode exceptions = response.path("exceptions"); + if (exceptions.isArray()) { + if (exceptions.isEmpty()) { + return false; + } + for (JsonNode exception : exceptions) { + if (!hasExpectedErrorMessage(exception, expectedErrorMessageContains)) { + return false; + } + } + return true; + } + return hasExpectedErrorMessage(exceptions, expectedErrorMessageContains); + } + + private static boolean hasExpectedErrorMessage(JsonNode exception, String expectedErrorMessageContains) { + JsonNode message = exception.path("message"); + return message.isTextual() && message.asText().contains(expectedErrorMessageContains); + } + private String getExplainPlan(String query) { try { if (!_useMultiStageQueryEngine) { JsonNode explainPlanResponse = - Utils.postSqlQuery("explain plan for " + query, ClusterDescriptor.getInstance().getBrokerUrl()); + Utils.postSqlQuery("explain plan for " + query, ClusterDescriptor.getInstance().getBrokerUrl(), + _enableNullHandling); return ExplainPlanUtils.formatExplainPlan(explainPlanResponse); } else { JsonNode explainPlanResponse = Utils.postMultiStageSqlQuery("explain plan for " + query, - ClusterDescriptor.getInstance().getBrokerUrl()); + ClusterDescriptor.getInstance().getBrokerUrl(), _enableNullHandling); return ExplainPlanUtils.formatMultiStageExplainPlan(explainPlanResponse); } } catch (Throwable error) { diff --git a/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/Utils.java b/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/Utils.java index edae1b17795..2713c8a9fe3 100644 --- a/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/Utils.java +++ b/pinot-compatibility-verifier/src/main/java/org/apache/pinot/compat/Utils.java @@ -50,9 +50,14 @@ public class Utils { public static JsonNode postSqlQuery(String query, String brokerBaseApiUrl) throws IOException { + return postSqlQuery(query, brokerBaseApiUrl, false); + } + + public static JsonNode postSqlQuery(String query, String brokerBaseApiUrl, boolean enableNullHandling) + throws IOException { ObjectNode payload = JsonUtils.newObjectNode(); payload.put("sql", query); - payload.put("queryOptions", "groupByMode=sql;responseFormat=sql"); + payload.put("queryOptions", getSingleStageQueryOptions(enableNullHandling)); return JsonUtils.stringToJsonNode( ControllerTest.sendPostRequest(brokerBaseApiUrl + "/query/sql", payload.toString())); @@ -60,11 +65,26 @@ public class Utils { public static JsonNode postMultiStageSqlQuery(String query, String brokerBaseApiUrl) throws IOException { + return postMultiStageSqlQuery(query, brokerBaseApiUrl, false); + } + + public static JsonNode postMultiStageSqlQuery(String query, String brokerBaseApiUrl, boolean enableNullHandling) + throws IOException { ObjectNode payload = JsonUtils.newObjectNode(); payload.put("sql", query); - payload.put("queryOptions", "useMultistageEngine=true"); + payload.put("queryOptions", getMultiStageQueryOptions(enableNullHandling)); return JsonUtils.stringToJsonNode( ControllerTest.sendPostRequest(brokerBaseApiUrl + "/query/sql", payload.toString())); } + + static String getSingleStageQueryOptions(boolean enableNullHandling) { + String queryOptions = "groupByMode=sql;responseFormat=sql"; + return enableNullHandling ? queryOptions + ";enableNullHandling=true" : queryOptions; + } + + static String getMultiStageQueryOptions(boolean enableNullHandling) { + String queryOptions = "useMultistageEngine=true"; + return enableNullHandling ? queryOptions + ";enableNullHandling=true" : queryOptions; + } } diff --git a/pinot-compatibility-verifier/src/test/java/org/apache/pinot/compat/BaseOpTest.java b/pinot-compatibility-verifier/src/test/java/org/apache/pinot/compat/BaseOpTest.java new file mode 100644 index 00000000000..b3e41a81d24 --- /dev/null +++ b/pinot-compatibility-verifier/src/test/java/org/apache/pinot/compat/BaseOpTest.java @@ -0,0 +1,78 @@ +/** + * 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. + */ +package org.apache.pinot.compat; + +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; + + +/// Tests capability-gated compatibility operations. +public class BaseOpTest { + private static final String TEST_PROPERTY = BaseOpTest.class.getName() + ".enabled"; + + @Test + public void testSystemPropertyRunCondition() { + String previousValue = System.getProperty(TEST_PROPERTY); + try { + CountingOp op = new CountingOp(); + assertTrue(op.run(1)); + assertEquals(op._runCount, 1); + + op.setRunIfSystemProperty(TEST_PROPERTY); + assertFalse(op.run(1)); + assertEquals(op._runCount, 1); + + op.setRunIfSystemPropertyValue("true"); + System.clearProperty(TEST_PROPERTY); + assertFalse(op.run(1)); + assertEquals(op._runCount, 1); + + System.setProperty(TEST_PROPERTY, "false"); + assertTrue(op.run(1)); + assertEquals(op._runCount, 1); + + System.setProperty(TEST_PROPERTY, "true"); + assertTrue(op.run(1)); + assertEquals(op._runCount, 2); + } finally { + if (previousValue == null) { + System.clearProperty(TEST_PROPERTY); + } else { + System.setProperty(TEST_PROPERTY, previousValue); + } + } + } + + private static final class CountingOp extends BaseOp { + private int _runCount; + + private CountingOp() { + super(OpType.QUERY_OP); + } + + @Override + boolean runOp(int generationNumber) { + _runCount++; + return true; + } + } +} diff --git a/pinot-compatibility-verifier/src/test/java/org/apache/pinot/compat/FileContainsOpTest.java b/pinot-compatibility-verifier/src/test/java/org/apache/pinot/compat/FileContainsOpTest.java new file mode 100644 index 00000000000..7c85172f570 --- /dev/null +++ b/pinot-compatibility-verifier/src/test/java/org/apache/pinot/compat/FileContainsOpTest.java @@ -0,0 +1,57 @@ +/** + * 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. + */ +package org.apache.pinot.compat; + +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; + + +/// Tests exact root-cause matching across compatibility-runner log files. +public class FileContainsOpTest { + + @Test + public void testContainsExpectedTextRequiresMatchingFileAndText() + throws Exception { + Path directory = Files.createTempDirectory("pinot-file-contains-op"); + Path serverLog = directory.resolve("server.4.log"); + Path brokerLog = directory.resolve("broker.3.log"); + try { + Files.writeString(serverLog, + "Caused by: java.lang.IllegalStateException: Unsupported proto ColumnDataType: UNRECOGNIZED\n", + StandardCharsets.UTF_8); + Files.writeString(brokerLog, "Caught exception while deserializing stage plan\n", StandardCharsets.UTF_8); + + assertTrue(FileContainsOp.containsExpectedText(directory, "server*.log", + "Unsupported proto ColumnDataType: UNRECOGNIZED")); + assertFalse(FileContainsOp.containsExpectedText(directory, "server*.log", + "Caught exception while deserializing stage plan")); + assertFalse(FileContainsOp.containsExpectedText(directory, "missing.*.log", + "Unsupported proto ColumnDataType: UNRECOGNIZED")); + } finally { + Files.deleteIfExists(serverLog); + Files.deleteIfExists(brokerLog); + Files.deleteIfExists(directory); + } + } +} diff --git a/pinot-compatibility-verifier/src/test/java/org/apache/pinot/compat/QueryOpTest.java b/pinot-compatibility-verifier/src/test/java/org/apache/pinot/compat/QueryOpTest.java new file mode 100644 index 00000000000..3a90d6024a3 --- /dev/null +++ b/pinot-compatibility-verifier/src/test/java/org/apache/pinot/compat/QueryOpTest.java @@ -0,0 +1,196 @@ +/** + * 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. + */ +package org.apache.pinot.compat; + +import com.fasterxml.jackson.databind.JsonNode; +import java.io.BufferedReader; +import java.io.IOException; +import java.io.StringReader; +import org.apache.pinot.spi.utils.JsonUtils; +import org.testng.annotations.Test; +import org.yaml.snakeyaml.DumperOptions; +import org.yaml.snakeyaml.LoaderOptions; +import org.yaml.snakeyaml.Yaml; +import org.yaml.snakeyaml.representer.Representer; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertThrows; +import static org.testng.Assert.assertTrue; + + +/// Tests expected query-error matching used by mixed-version compatibility suites. +public class QueryOpTest { + + @Test + public void testMatchesExpectedErrorInExceptionArray() + throws Exception { + JsonNode response = JsonUtils.stringToJsonNode( + "{\"exceptions\":[" + + "{\"errorCode\":200,\"message\":\"java.lang.IllegalStateException: " + + "Unsupported proto ColumnDataType: UNRECOGNIZED\"}," + + "{\"errorCode\":200,\"message\":\"Unsupported proto ColumnDataType: UNRECOGNIZED\"}]}"); + + assertTrue(QueryOp.hasExpectedError(response, "Unsupported proto ColumnDataType: UNRECOGNIZED")); + assertFalse(QueryOp.hasExpectedError(response, "Unsupported proto ColumnDataType: VARIANT")); + + JsonNode mixedResponse = JsonUtils.stringToJsonNode( + "{\"exceptions\":[" + + "{\"errorCode\":200,\"message\":\"Unsupported proto ColumnDataType: UNRECOGNIZED\"}," + + "{\"errorCode\":200,\"message\":\"Server request timed out\"}]}"); + assertFalse(QueryOp.hasExpectedError(mixedResponse, "Unsupported proto ColumnDataType: UNRECOGNIZED")); + + JsonNode partialRowsResponse = JsonUtils.stringToJsonNode( + "{\"resultTable\":{\"rows\":[[10]]},\"numRowsResultSet\":1," + + "\"exceptions\":[{\"message\":\"Unsupported proto ColumnDataType: UNRECOGNIZED\"}]}"); + assertFalse(QueryOp.hasExpectedError(partialRowsResponse, "Unsupported proto ColumnDataType: UNRECOGNIZED")); + } + + @Test + public void testMatchesExpectedErrorInExceptionObject() + throws Exception { + JsonNode response = + JsonUtils.stringToJsonNode("{\"exceptions\":{\"message\":\"Unsupported proto ColumnDataType: UNRECOGNIZED\"}}"); + + assertTrue(QueryOp.hasExpectedError(response, "ColumnDataType: UNRECOGNIZED")); + } + + @Test + public void testDoesNotMatchMissingOrMalformedExceptions() + throws Exception { + String expectedMessage = "Unsupported proto ColumnDataType: UNRECOGNIZED"; + assertFalse(QueryOp.hasExpectedError(null, expectedMessage)); + assertFalse(QueryOp.hasExpectedError(JsonUtils.stringToJsonNode("{}"), expectedMessage)); + assertFalse(QueryOp.hasExpectedError(JsonUtils.stringToJsonNode("{\"exceptions\":[]}"), + expectedMessage)); + assertFalse(QueryOp.hasExpectedError( + JsonUtils.stringToJsonNode("{\"exceptions\":[{\"errorCode\":200}]}"), expectedMessage)); + assertFalse(QueryOp.hasExpectedError( + JsonUtils.stringToJsonNode("{\"exceptions\":[{\"message\":42}]}"), expectedMessage)); + assertFalse(QueryOp.hasExpectedError( + JsonUtils.stringToJsonNode("{\"exceptions\":[{\"message\":\"" + expectedMessage + "\"}]}"), "")); + } + + @Test + public void testExpectedOutcomeConfigurationRequiresExactlyOneMode() { + QueryOp queryOp = new QueryOp(); + assertFalse(queryOp.hasValidExpectedOutcomeConfiguration()); + + queryOp.setExpectedResultsFileName("query-results/results.json"); + assertTrue(queryOp.hasValidExpectedOutcomeConfiguration()); + + queryOp.setExpectedErrorMessageContains(" "); + assertFalse(queryOp.hasValidExpectedOutcomeConfiguration()); + + queryOp.setExpectedErrorMessageContains("Unsupported proto ColumnDataType: UNRECOGNIZED"); + assertFalse(queryOp.hasValidExpectedOutcomeConfiguration()); + + queryOp.setExpectedResultsFileName(null); + assertTrue(queryOp.hasValidExpectedOutcomeConfiguration()); + + queryOp.setExpectedErrorMessageContains(" "); + assertFalse(queryOp.hasValidExpectedOutcomeConfiguration()); + } + + @Test + public void testRetriesAreExplicitAndBounded() { + int[] attempts = {0}; + QueryOp queryOp = new QueryOp() { + @Override + boolean verifyQueries(int generationNumber) { + return ++attempts[0] == 3; + } + }; + queryOp.setQueryFileName("queries/routing-ready.queries"); + queryOp.setExpectedResultsFileName("query-results/variant-wire.results"); + queryOp.setMaxAttempts(3); + queryOp.setRetryDelayMs(0); + + assertTrue(queryOp.runOp(1)); + assertEquals(attempts[0], 3); + + queryOp.setMaxAttempts(0); + assertFalse(queryOp.runOp(1)); + queryOp.setMaxAttempts(1); + queryOp.setRetryDelayMs(-1); + assertFalse(queryOp.runOp(1)); + } + + @Test + public void testExpectedResultReaderReportsTruncatedFiles() + throws IOException { + BufferedReader reader = new BufferedReader(new StringReader("# comment\n\n{\"resultTable\":{}}\n")); + assertEquals(QueryOp.readNextExpectedResult(reader, 21), "{\"resultTable\":{}}"); + assertThrows(IOException.class, () -> QueryOp.readNextExpectedResult(reader, 22)); + } + + @Test + public void testNormalResultComparisonUsesSelectedEngine() + throws Exception { + JsonNode actualSubset = successfulResponse(1, 1); + JsonNode expectedSuperset = JsonUtils.stringToJsonNode( + "{\"resultTable\":{\"dataSchema\":{\"columnNames\":[\"EXPR$0\"],\"columnDataTypes\":[\"LONG\"]}," + + "\"rows\":[[1],[2]]},\"exceptions\":[],\"numDocsScanned\":1,\"isSuperset\":true}"); + String query = "SELECT COUNT(*) FROM testTable"; + + assertTrue(QueryOp.matchesExpectedResponse(actualSubset, expectedSuperset, null, false, query)); + assertFalse(QueryOp.matchesExpectedResponse(actualSubset, expectedSuperset, null, true, query)); + + JsonNode moreExpensiveActual = successfulResponse(1, 2); + JsonNode expected = successfulResponse(1, 1); + assertFalse(QueryOp.matchesExpectedResponse(moreExpensiveActual, expected, null, false, query)); + assertTrue(QueryOp.matchesExpectedResponse(moreExpensiveActual, expected, null, true, query)); + } + + @Test + public void testNullHandlingQueryOptionsAreOptIn() { + assertEquals(Utils.getSingleStageQueryOptions(false), "groupByMode=sql;responseFormat=sql"); + assertEquals(Utils.getSingleStageQueryOptions(true), + "groupByMode=sql;responseFormat=sql;enableNullHandling=true"); + assertEquals(Utils.getMultiStageQueryOptions(false), "useMultistageEngine=true"); + assertEquals(Utils.getMultiStageQueryOptions(true), "useMultistageEngine=true;enableNullHandling=true"); + } + + @Test + public void testNullHandlingYamlPropertyUsesBooleanNaming() { + DumperOptions dumperOptions = new DumperOptions(); + Representer representer = new Representer(dumperOptions); + representer.getPropertyUtils().setSkipMissingProperties(true); + Yaml yaml = new Yaml(new CompatibilityOpsRunner.CustomConstructor(new LoaderOptions()), representer, dumperOptions); + + CompatTestOperation operation = yaml.loadAs(""" + description: Test null handling property + operations: + - type: queryOp + queryFileName: queries/test.queries + expectedResultsFileName: query-results/test.results + nullHandlingEnabled: true + """, CompatTestOperation.class); + QueryOp queryOp = (QueryOp) operation.getOperations().get(0); + + assertTrue(queryOp.isNullHandlingEnabled()); + } + + private static JsonNode successfulResponse(int value, int numDocsScanned) + throws Exception { + return JsonUtils.stringToJsonNode( + "{\"resultTable\":{\"dataSchema\":{\"columnNames\":[\"EXPR$0\"],\"columnDataTypes\":[\"LONG\"]}," + + "\"rows\":[[" + value + "]]},\"exceptions\":[],\"numDocsScanned\":" + numDocsScanned + "}"); + } +} diff --git a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/ClusterTest.java b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/ClusterTest.java index 4ce8b26ef99..c60d1caa6a2 100644 --- a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/ClusterTest.java +++ b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/ClusterTest.java @@ -788,6 +788,7 @@ public abstract class ClusterTest extends ControllerTest { case STRING: case BYTES: case JSON: + case VARIANT: object = jsonValue.textValue(); break; case UNKNOWN: diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/VariantTypeTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/VariantTypeTest.java new file mode 100644 index 00000000000..483bca28e5f --- /dev/null +++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/VariantTypeTest.java @@ -0,0 +1,661 @@ +/** + * 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. + */ +package org.apache.pinot.integration.tests.custom; + +import com.fasterxml.jackson.databind.JsonNode; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.StandardCopyOption; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Set; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.example.ExampleParquetWriter; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; +import org.apache.pinot.integration.tests.ClusterIntegrationTestUtils; +import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader; +import org.apache.pinot.segment.spi.ImmutableSegment; +import org.apache.pinot.segment.spi.index.reader.ForwardIndexReader; +import org.apache.pinot.segment.spi.index.reader.ForwardIndexReaderContext; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.data.readers.FileFormat; +import org.apache.pinot.spi.utils.JsonUtils; +import org.apache.pinot.spi.utils.ReadMode; +import org.apache.pinot.spi.utils.VariantEnvelope; +import org.testng.Assert; +import org.testng.annotations.Test; + + +/// End-to-end coverage for creating a VARIANT table, ingesting Parquet VARIANT(1), materializing a hot path, and +/// querying nested values with both Pinot query engines. +@Test(suiteName = "CustomClusterIntegrationTest") +public class VariantTypeTest extends CustomDataQueryClusterIntegrationTest { + private static final String RESOURCE_DIRECTORY = "examples/batch/variantEvents/"; + private static final String TABLE_NAME = "variantEvents"; + private static final String EVENT_ID = "eventId"; + private static final String EVENT_TYPE = "eventType"; + private static final String PAYLOAD = "payload"; + private static final int NUM_DOCS = 6; + + @Override + public String getTableName() { + return TABLE_NAME; + } + + @Override + protected long getCountStarResult() { + return NUM_DOCS; + } + + @Override + public Schema createSchema() { + try (InputStream inputStream = openResource("variantEvents_schema.json")) { + return Schema.fromInputStream(inputStream); + } catch (IOException e) { + throw new IllegalStateException("Failed to load the VARIANT quickstart schema", e); + } + } + + @Override + public TableConfig createOfflineTableConfig() { + try (InputStream inputStream = openResource("variantEvents_offline_table_config.json")) { + return JsonUtils.inputStreamToObject(inputStream, TableConfig.class); + } catch (IOException e) { + throw new IllegalStateException("Failed to load the VARIANT quickstart table config", e); + } + } + + @Override + protected void setUpTable() + throws Exception { + Schema schema = createSchema(); + addSchema(schema); + TableConfig tableConfig = createOfflineTableConfig(); + addTableConfig(tableConfig); + + File parquetFile = new File(_tempDir, "variantEvents_data.parquet"); + try (InputStream inputStream = openResource("rawdata/variantEvents_data.parquet")) { + Files.copy(inputStream, parquetFile.toPath(), StandardCopyOption.REPLACE_EXISTING); + } + ClusterIntegrationTestUtils.buildSegmentFromFile(parquetFile, tableConfig, schema, "0", _segmentDir, _tarDir, + FileFormat.PARQUET); + + Set<String> existingSegments = segmentNames(_segmentDir); + byte[] nonMonotonicEnvelope = nonMonotonicObjectEnvelope(); + File interoperabilityFile = writeNonMonotonicVariantFile(nonMonotonicEnvelope); + ClusterIntegrationTestUtils.buildSegmentFromFile(interoperabilityFile, tableConfig, schema, "1", _segmentDir, + _tarDir, FileFormat.PARQUET); + File interoperabilitySegment = findNewSegment(_segmentDir, existingSegments); + assertSegmentPreservesVariantEnvelope(interoperabilitySegment, nonMonotonicEnvelope); + uploadSegments(getTableName(), _tarDir); + } + + @Override + public List<File> createAvroFiles() { + throw new UnsupportedOperationException("VariantTypeTest ingests Parquet, not Avro"); + } + + @Test(dataProvider = "useBothQueryEngines") + public void testMaterializedPathAndDirectExtraction(boolean useMultiStageQueryEngine) + throws Exception { + setUseMultiStageQueryEngine(useMultiStageQueryEngine); + + JsonNode response = postVariantQuery( + "SELECT " + EVENT_ID + ", " + EVENT_TYPE + " FROM " + TABLE_NAME + + " WHERE " + EVENT_TYPE + " = 'checkout' ORDER BY " + EVENT_ID); + assertNoExceptions(response); + JsonNode rows = response.get("resultTable").get("rows"); + Assert.assertEquals( + response.get("resultTable").get("dataSchema").get("columnDataTypes").toString(), + "[\"STRING\",\"STRING\"]"); + Assert.assertEquals(rows.size(), 2); + Assert.assertEquals(rows.get(0).get(0).asText(), "evt-001"); + Assert.assertEquals(rows.get(0).get(1).asText(), "checkout"); + Assert.assertEquals(rows.get(1).get(0).asText(), "evt-003"); + Assert.assertEquals(rows.get(1).get(1).asText(), "checkout"); + + response = postVariantQuery( + "SELECT " + EVENT_ID + ", variantGet(" + PAYLOAD + ", '$.user.id', 'STRING'), " + + "variantGet(" + PAYLOAD + ", '$.amount', 'DOUBLE') FROM " + TABLE_NAME + + " WHERE " + EVENT_TYPE + " = 'checkout' ORDER BY " + EVENT_ID); + assertNoExceptions(response); + Assert.assertEquals( + response.get("resultTable").get("dataSchema").get("columnDataTypes").toString(), + "[\"STRING\",\"STRING\",\"DOUBLE\"]"); + rows = response.get("resultTable").get("rows"); + Assert.assertEquals(rows.size(), 2); + Assert.assertEquals(rows.get(0).get(0).asText(), "evt-001"); + Assert.assertEquals(rows.get(0).get(1).asText(), "u-1"); + Assert.assertEquals(rows.get(0).get(2).asDouble(), 42.5); + Assert.assertEquals(rows.get(1).get(0).asText(), "evt-003"); + Assert.assertEquals(rows.get(1).get(1).asText(), "u-3"); + Assert.assertEquals(rows.get(1).get(2).asDouble(), 19.0); + + if (!useMultiStageQueryEngine) { + Assert.assertEquals(response.get("numEntriesScannedInFilter").asLong(), 0L, + "The materialized eventType filter should use its inverted index"); + } + } + + @Test(dataProvider = "useBothQueryEngines") + public void testNonMonotonicObjectOffsetsFromExternalParquet(boolean useMultiStageQueryEngine) + throws Exception { + setUseMultiStageQueryEngine(useMultiStageQueryEngine); + + JsonNode response = postVariantQuery( + "SELECT variant_get(" + PAYLOAD + ", '$.a', 'INT'), " + + "try_variant_get(" + PAYLOAD + ", '$.b', 'INT'), " + + "variant_get(" + PAYLOAD + ", '$.c'), variant_to_json(" + PAYLOAD + ") " + + "FROM " + TABLE_NAME + " WHERE " + EVENT_ID + " = 'evt-006'"); + assertNoExceptions(response); + Assert.assertEquals( + response.get("resultTable").get("dataSchema").get("columnDataTypes").toString(), + "[\"INT\",\"INT\",\"VARIANT\",\"STRING\"]"); + JsonNode row = response.get("resultTable").get("rows").get(0); + Assert.assertEquals(row.get(0).asInt(), 1); + Assert.assertEquals(row.get(1).asInt(), 2); + Assert.assertEquals(row.get(2).asText(), "3"); + Assert.assertEquals(row.get(3).asText(), "{\"a\":1,\"b\":2,\"c\":3,\"eventType\":\"interop\"}"); + + response = postVariantQuery( + "SELECT try_variant_get(" + PAYLOAD + ", '$.missing', 'INT') FROM " + TABLE_NAME + + " WHERE " + EVENT_ID + " = 'evt-006'"); + assertNoExceptions(response); + Assert.assertTrue(response.get("resultTable").get("rows").get(0).get(0).isNull()); + } + + @Test(dataProvider = "useBothQueryEngines") + public void testJsonProjectionAndNullStates(boolean useMultiStageQueryEngine) + throws Exception { + setUseMultiStageQueryEngine(useMultiStageQueryEngine); + + JsonNode response = postVariantQuery( + "SELECT " + PAYLOAD + ", variantToJson(" + PAYLOAD + ") FROM " + TABLE_NAME + + " WHERE " + EVENT_ID + " = 'evt-001'"); + assertNoExceptions(response); + Assert.assertEquals( + response.get("resultTable").get("dataSchema").get("columnDataTypes").toString(), + "[\"VARIANT\",\"STRING\"]"); + JsonNode row = response.get("resultTable").get("rows").get(0); + String json = row.get(0).asText(); + Assert.assertEquals(row.get(1).asText(), json); + Assert.assertEquals(json, + "{\"amount\":42.5,\"eventType\":\"checkout\",\"items\":[\"sku-1\",\"sku-2\"]," + + "\"user\":{\"id\":\"u-1\"}}"); + JsonNode payload = JsonUtils.stringToJsonNode(json); + Assert.assertEquals(payload.get("eventType").asText(), "checkout"); + Assert.assertEquals(payload.get("user").get("id").asText(), "u-1"); + Assert.assertEquals(payload.get("items").size(), 2); + + response = postVariantQuery( + "SELECT " + EVENT_ID + ", variantExists(" + PAYLOAD + ", '$.coupon'), " + + "isVariantNull(" + PAYLOAD + ", '$.coupon') FROM " + TABLE_NAME + + " WHERE " + EVENT_ID + " IN ('evt-001', 'evt-002', 'evt-003') ORDER BY " + EVENT_ID); + assertNoExceptions(response); + JsonNode rows = response.get("resultTable").get("rows"); + Assert.assertEquals(rows.size(), 3); + Assert.assertFalse(rows.get(0).get(1).asBoolean()); + Assert.assertFalse(rows.get(0).get(2).asBoolean()); + Assert.assertTrue(rows.get(1).get(1).asBoolean()); + Assert.assertTrue(rows.get(1).get(2).asBoolean()); + Assert.assertTrue(rows.get(2).get(1).asBoolean()); + Assert.assertFalse(rows.get(2).get(2).asBoolean()); + + response = postVariantQuery( + "SELECT " + EVENT_ID + ", variantTypeOf(" + PAYLOAD + ", '$') FROM " + TABLE_NAME + + " WHERE " + EVENT_ID + " IN ('evt-004', 'evt-005') ORDER BY " + EVENT_ID); + assertNoExceptions(response); + rows = response.get("resultTable").get("rows"); + Assert.assertEquals(rows.size(), 2); + Assert.assertEquals(rows.get(0).get(1).asText(), "NULL"); + Assert.assertTrue(rows.get(1).get(1).isNull()); + + response = postVariantQuery( + "SELECT " + EVENT_ID + ", " + PAYLOAD + " FROM " + TABLE_NAME + + " WHERE " + EVENT_ID + " IN ('evt-004', 'evt-005') ORDER BY " + EVENT_ID); + assertNoExceptions(response); + Assert.assertEquals( + response.get("resultTable").get("dataSchema").get("columnDataTypes").toString(), + "[\"STRING\",\"VARIANT\"]"); + rows = response.get("resultTable").get("rows"); + Assert.assertEquals(rows.get(0).get(1).asText(), "null", + "An encoded Variant null must render as JSON text"); + Assert.assertTrue(rows.get(1).get(1).isNull(), "A missing Parquet payload must remain SQL null"); + } + + @Test(dataProvider = "useBothQueryEngines") + public void testSparkCompatibleFunctionSemantics(boolean useMultiStageQueryEngine) + throws Exception { + setUseMultiStageQueryEngine(useMultiStageQueryEngine); + + JsonNode response = postVariantQuery( + "SELECT variant_get(" + PAYLOAD + ", '$.user'), " + + "variant_get(" + PAYLOAD + ", '$.missing'), " + + "variant_get(" + PAYLOAD + ", '$.items[1]', 'STRING'), " + + "try_variant_get(" + PAYLOAD + ", '$.eventType', 'DOUBLE') " + + "FROM " + TABLE_NAME + " WHERE " + EVENT_ID + " = 'evt-001'"); + assertNoExceptions(response); + Assert.assertEquals( + response.get("resultTable").get("dataSchema").get("columnDataTypes").toString(), + "[\"VARIANT\",\"VARIANT\",\"STRING\",\"DOUBLE\"]"); + JsonNode row = response.get("resultTable").get("rows").get(0); + Assert.assertEquals(row.get(0).asText(), "{\"id\":\"u-1\"}"); + Assert.assertTrue(row.get(1).isNull(), "A missing path must return SQL null"); + Assert.assertEquals(row.get(2).asText(), "sku-2"); + Assert.assertTrue(row.get(3).isNull(), "try_variant_get must return SQL null for an incompatible cast"); + + response = postVariantQuery( + "SELECT variant_get(parse_json('{\"answer\":42}'), '$.answer', 'INT') " + + "FROM " + TABLE_NAME + " LIMIT 1"); + assertNoExceptions(response); + Assert.assertEquals(response.get("resultTable").get("rows").get(0).get(0).asInt(), 42); + + response = postVariantQuery( + "SELECT is_variant_null(" + PAYLOAD + ") FROM " + TABLE_NAME + + " WHERE " + EVENT_ID + " = 'evt-005'"); + assertNoExceptions(response); + JsonNode isVariantNull = response.get("resultTable").get("rows").get(0).get(0); + Assert.assertFalse(isVariantNull.isNull(), "is_variant_null(SQL NULL) must return a non-null boolean"); + Assert.assertFalse(isVariantNull.asBoolean(), + "SQL null is not an encoded Variant null"); + + response = postVariantQuery( + "SELECT variant_get(" + PAYLOAD + ", '$.eventType', 'DOUBLE') FROM " + TABLE_NAME + + " WHERE " + EVENT_ID + " = 'evt-001'"); + assertExceptionContains(response, "cannot convert variant", "double"); + } + + @Test(dataProvider = "useBothQueryEngines") + public void testRawVariantInAndNotInAreRejected(boolean useMultiStageQueryEngine) + throws Exception { + setUseMultiStageQueryEngine(useMultiStageQueryEngine); + + for (String operator : List.of("IN", "NOT IN")) { + JsonNode response = postVariantQuery( + "SELECT " + EVENT_ID + " FROM " + TABLE_NAME + " WHERE " + PAYLOAD + " " + operator + + " (parse_json('{\"candidate\":1}'), parse_json('{\"candidate\":2}'))"); + if (useMultiStageQueryEngine) { + assertExceptionContains(response, "raw variant", "in"); + } else { + // Single-stage no longer constant-folds parse_json (the fold would erase the VARIANT type), so the IN list + // has no literal values and the query is rejected at parsing rather than by the runtime capability guard. + assertHasException(response); + } + } + } + + @Test(dataProvider = "useBothQueryEngines") + public void testRawVariantRegexpLikeIsRejected(boolean useMultiStageQueryEngine) + throws Exception { + setUseMultiStageQueryEngine(useMultiStageQueryEngine); + + JsonNode response = postVariantQuery( + "SELECT " + EVENT_ID + " FROM " + TABLE_NAME + " WHERE REGEXP_LIKE(" + PAYLOAD + ", '.*')"); + // Calcite inserts a VARCHAR cast for REGEXP_LIKE in the multi-stage plan, so that engine rejects the raw + // VARIANT at the earlier CAST boundary. The single-stage engine reaches the predicate-specific guard. + assertExceptionContains(response, "raw variant", useMultiStageQueryEngine ? "cast" : "regexp_like"); + } + + @Test(dataProvider = "useBothQueryEngines") + public void testRawVariantOrderByIsRejected(boolean useMultiStageQueryEngine) + throws Exception { + setUseMultiStageQueryEngine(useMultiStageQueryEngine); + + JsonNode response = + postVariantQuery("SELECT " + EVENT_ID + " FROM " + TABLE_NAME + " ORDER BY " + PAYLOAD + " LIMIT 5"); + assertExceptionContains(response, "raw variant", "order by"); + } + + @Test(dataProvider = "useBothQueryEngines") + public void testRawVariantComparisonGroupingAndDistinctAreRejected(boolean useMultiStageQueryEngine) + throws Exception { + setUseMultiStageQueryEngine(useMultiStageQueryEngine); + + JsonNode response = postVariantQuery( + "SELECT " + EVENT_ID + " FROM " + TABLE_NAME + " WHERE " + PAYLOAD + + " = parse_json('{\"eventType\":\"checkout\"}')"); + assertExceptionContains(response, "raw variant", "comparison"); + for (String operator : List.of("IS DISTINCT FROM", "IS NOT DISTINCT FROM")) { + response = postVariantQuery( + "SELECT " + EVENT_ID + " FROM " + TABLE_NAME + " WHERE " + PAYLOAD + " " + operator + + " parse_json(variant_to_json(" + PAYLOAD + "))"); + assertExceptionContains(response, "raw variant", "comparison"); + } + response = postVariantQuery( + "SELECT " + PAYLOAD + ", COUNT(*) FROM " + TABLE_NAME + " GROUP BY " + PAYLOAD); + assertExceptionContains(response, "raw variant", "group by"); + // A filtered aggregation routes to the filtered group-by operator, whose guard is otherwise unreachable. + response = postVariantQuery( + "SELECT " + PAYLOAD + ", COUNT(*) FILTER (WHERE " + EVENT_TYPE + " = 'checkout') FROM " + TABLE_NAME + + " GROUP BY " + PAYLOAD); + assertExceptionContains(response, "raw variant", "group by"); + // Scalar functions and COALESCE must not consume the raw envelope as text. Multi-stage inserts an implicit CAST + // that its own guard (or Calcite) rejects, so only the single-stage message is pinned. + for (String query : List.of( + "SELECT LENGTH(" + PAYLOAD + ") FROM " + TABLE_NAME, + "SELECT COALESCE(" + PAYLOAD + ", 'x') FROM " + TABLE_NAME)) { + response = postVariantQuery(query); + if (useMultiStageQueryEngine) { + assertHasException(response); + } else { + assertExceptionContains(response, "raw variant"); + } + } + response = postVariantQuery("SELECT DISTINCT " + PAYLOAD + " FROM " + TABLE_NAME); + assertExceptionContains(response, "raw variant", useMultiStageQueryEngine ? "group by" : "distinct"); + + for (String query : List.of( + "SELECT " + EVENT_ID + " FROM " + TABLE_NAME + " WHERE variant_get(" + PAYLOAD + + ", '$.eventType', 'STRING') = 'checkout'", + "SELECT variant_get(" + PAYLOAD + ", '$.eventType', 'STRING'), COUNT(*) FROM " + TABLE_NAME + + " GROUP BY variant_get(" + PAYLOAD + ", '$.eventType', 'STRING')", + "SELECT variant_get(" + PAYLOAD + ", '$.eventType', 'STRING'), COUNT(*) FILTER (WHERE " + EVENT_TYPE + + " = 'checkout') FROM " + TABLE_NAME + " GROUP BY variant_get(" + PAYLOAD + ", '$.eventType', 'STRING')", + "SELECT LENGTH(variant_get(" + PAYLOAD + ", '$.eventType', 'STRING')) FROM " + TABLE_NAME, + "SELECT DISTINCT variant_get(" + PAYLOAD + ", '$.eventType', 'STRING') FROM " + TABLE_NAME)) { + assertNoExceptions(postVariantQuery(query)); + } + } + + /// CAST would leak the raw PVAR envelope as STRING/JSON/BYTES hex, bypassing the opacity contract; both engines + /// must reject it while CAST over an extracted typed path keeps working. + @Test(dataProvider = "useBothQueryEngines") + public void testRawVariantCastIsRejected(boolean useMultiStageQueryEngine) + throws Exception { + setUseMultiStageQueryEngine(useMultiStageQueryEngine); + for (String targetType : List.of("VARCHAR", "VARBINARY", "JSON")) { + JsonNode response = postVariantQuery( + "SELECT CAST(" + PAYLOAD + " AS " + targetType + ") FROM " + TABLE_NAME); + if (useMultiStageQueryEngine) { + // Calcite's native VARIANT type may reject the cast at validation ("Cannot cast ...") before the runtime + // guard is reached; either rejection keeps the envelope opaque. + assertHasException(response); + } else { + assertExceptionContains(response, "raw variant", "cast"); + } + } + assertNoExceptions(postVariantQuery( + "SELECT CAST(variant_get(" + PAYLOAD + ", '$.eventType', 'STRING') AS VARCHAR) FROM " + TABLE_NAME)); + } + + @Test(dataProvider = "useBothQueryEngines") + public void testRawVariantAggregatesAreRejected(boolean useMultiStageQueryEngine) + throws Exception { + setUseMultiStageQueryEngine(useMultiStageQueryEngine); + + for (String aggregate : List.of("DISTINCTCOUNT", "DISTINCTCOUNTHLL", "DISTINCTCOUNTTHETASKETCH")) { + JsonNode response = postVariantQuery("SELECT " + aggregate + "(" + PAYLOAD + ") FROM " + TABLE_NAME); + assertExceptionContains(response, "raw variant", aggregate); + } + + JsonNode response = postVariantQuery("SELECT COUNT(" + PAYLOAD + ") FROM " + TABLE_NAME); + assertNoExceptions(response); + Assert.assertEquals(response.get("resultTable").get("rows").get(0).get(0).asLong(), 5L, + "COUNT is raw-value-independent and must retain SQL-null semantics"); + } + + @Test(dataProvider = "useBothQueryEngines") + public void testVariantFunctionsRequireQueryNullHandling(boolean useMultiStageQueryEngine) + throws Exception { + setUseMultiStageQueryEngine(useMultiStageQueryEngine); + + JsonNode response = postQuery( + "SELECT variant_get(" + PAYLOAD + ", '$.eventType', 'STRING') FROM " + TABLE_NAME + " LIMIT 1"); + assertExceptionContains(response, "requires query null handling"); + } + + @Test(dataProvider = "useBothQueryEngines") + public void testRawVariantProjectionRequiresQueryNullHandling(boolean useMultiStageQueryEngine) + throws Exception { + setUseMultiStageQueryEngine(useMultiStageQueryEngine); + + JsonNode response = postQuery( + "SELECT " + PAYLOAD + " FROM " + TABLE_NAME + + " WHERE " + EVENT_ID + " IN ('evt-001', 'evt-005') ORDER BY " + EVENT_ID); + assertExceptionContains(response, "raw variant", "requires query null handling"); + + if (!useMultiStageQueryEngine) { + for (String predicate : List.of("1 = 0", "eventTime < 0")) { + response = postQuery("SELECT " + PAYLOAD + " FROM " + TABLE_NAME + " WHERE " + predicate); + assertExceptionContains(response, "raw variant", "requires query null handling"); + } + } + } + + @Test + public void testRawVariantJoinIsRejectedButTypedPathJoinWorks() + throws Exception { + setUseMultiStageQueryEngine(true); + String rawJoin = "SELECT leftTable." + EVENT_ID + " FROM " + TABLE_NAME + " leftTable JOIN " + TABLE_NAME + + " rightTable ON leftTable." + PAYLOAD + " = rightTable." + PAYLOAD; + JsonNode response = postVariantQuery(rawJoin); + assertExceptionContains(response, "raw variant", "join"); + + String typedJoin = "SELECT leftTable." + EVENT_ID + ", rightTable." + EVENT_ID + " FROM " + TABLE_NAME + + " leftTable JOIN " + TABLE_NAME + " rightTable ON variant_get(leftTable." + PAYLOAD + + ", '$.eventType', 'STRING') = variant_get(rightTable." + PAYLOAD + ", '$.eventType', 'STRING')" + + " WHERE leftTable." + EVENT_ID + " = 'evt-001' AND rightTable." + EVENT_ID + " = 'evt-003'"; + response = postVariantQuery(typedJoin); + assertNoExceptions(response); + Assert.assertEquals(response.get("resultTable").get("rows").size(), 1); + Assert.assertEquals(response.get("resultTable").get("rows").get(0).get(0).asText(), "evt-001"); + Assert.assertEquals(response.get("resultTable").get("rows").get(0).get(1).asText(), "evt-003"); + } + + @Test + public void testRawVariantWindowKeysAreRejectedButTypedPathWorks() + throws Exception { + setUseMultiStageQueryEngine(true); + + for (String window : List.of( + "COUNT(*) OVER (PARTITION BY " + PAYLOAD + ")", + "COUNT(*) OVER (ORDER BY " + PAYLOAD + ")")) { + JsonNode response = postVariantQuery("SELECT " + window + " FROM " + TABLE_NAME); + assertExceptionContains(response, "raw variant", "window"); + } + + JsonNode response = postVariantQuery( + "SELECT " + EVENT_ID + ", variant_get(" + PAYLOAD + ", '$.eventType', 'STRING'), " + + "COUNT(*) OVER (PARTITION BY variant_get(" + PAYLOAD + + ", '$.eventType', 'STRING')) FROM " + TABLE_NAME + " ORDER BY " + EVENT_ID); + assertNoExceptions(response); + Assert.assertEquals( + response.get("resultTable").get("dataSchema").get("columnDataTypes").toString(), + "[\"STRING\",\"STRING\",\"LONG\"]"); + JsonNode rows = response.get("resultTable").get("rows"); + Assert.assertEquals(rows.size(), NUM_DOCS); + Assert.assertEquals(rows.get(0).get(0).asText(), "evt-001"); + Assert.assertEquals(rows.get(0).get(1).asText(), "checkout"); + Assert.assertEquals(rows.get(0).get(2).asLong(), 2L); + Assert.assertEquals(rows.get(1).get(0).asText(), "evt-002"); + Assert.assertEquals(rows.get(1).get(1).asText(), "view"); + Assert.assertEquals(rows.get(1).get(2).asLong(), 1L); + Assert.assertEquals(rows.get(2).get(0).asText(), "evt-003"); + Assert.assertEquals(rows.get(2).get(1).asText(), "checkout"); + Assert.assertEquals(rows.get(2).get(2).asLong(), 2L); + Assert.assertEquals(rows.get(3).get(0).asText(), "evt-004"); + Assert.assertTrue(rows.get(3).get(1).isNull(), "Variant null must extract to the SQL-null partition"); + Assert.assertEquals(rows.get(3).get(2).asLong(), 2L); + Assert.assertEquals(rows.get(4).get(0).asText(), "evt-005"); + Assert.assertTrue(rows.get(4).get(1).isNull(), "SQL null must remain in the SQL-null partition"); + Assert.assertEquals(rows.get(4).get(2).asLong(), 2L); + Assert.assertEquals(rows.get(5).get(0).asText(), "evt-006"); + Assert.assertEquals(rows.get(5).get(1).asText(), "interop"); + Assert.assertEquals(rows.get(5).get(2).asLong(), 1L); + } + + @Test + public void testRawVariantSetOperationsAreRejected() + throws Exception { + setUseMultiStageQueryEngine(true); + String left = "SELECT " + PAYLOAD + " FROM " + TABLE_NAME + " WHERE " + EVENT_ID + " = 'evt-001'"; + String right = "SELECT " + PAYLOAD + " FROM " + TABLE_NAME + " WHERE " + EVENT_ID + " = 'evt-002'"; + + for (String operator : List.of("UNION", "INTERSECT", "INTERSECT ALL", "EXCEPT", "EXCEPT ALL")) { + JsonNode response = postVariantQuery(left + " " + operator + " " + right); + assertExceptionContains(response, "raw variant", "extract a typed path"); + } + + JsonNode response = postVariantQuery(left + " UNION ALL " + right); + assertNoExceptions(response); + Assert.assertEquals( + response.get("resultTable").get("dataSchema").get("columnDataTypes").toString(), "[\"VARIANT\"]"); + Assert.assertEquals(response.get("resultTable").get("rows").size(), 2); + } + + private File writeNonMonotonicVariantFile(byte[] envelope) + throws IOException { + MessageType parquetSchema = MessageTypeParser.parseMessageType( + "message variant_interoperability {" + + " required binary eventId (STRING);" + + " required int64 eventTime;" + + " optional group payload (VARIANT(1)) {" + + " required binary metadata;" + + " required binary value;" + + " }" + + "}"); + File parquetFile = new File(_tempDir, "variant_non_monotonic_offsets.parquet"); + Files.deleteIfExists(parquetFile.toPath()); + VariantEnvelope.Decoded decoded = VariantEnvelope.decode(envelope); + try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(new Path(parquetFile.getAbsolutePath())) + .withType(parquetSchema).build()) { + Group row = new SimpleGroupFactory(parquetSchema).newGroup() + .append(EVENT_ID, "evt-006") + .append("eventTime", 1_700_000_005_000L); + row.addGroup(PAYLOAD) + .append("metadata", Binary.fromConstantByteBuffer(decoded.getMetadata())) + .append("value", Binary.fromConstantByteBuffer(decoded.getValue())); + writer.write(row); + } + return parquetFile; + } + + private static Set<String> segmentNames(File segmentDirectory) { + Set<String> names = new HashSet<>(); + File[] segments = segmentDirectory.listFiles(File::isDirectory); + if (segments != null) { + for (File segment : segments) { + names.add(segment.getName()); + } + } + return names; + } + + private static File findNewSegment(File segmentDirectory, Set<String> existingNames) { + File[] segments = segmentDirectory.listFiles(File::isDirectory); + Assert.assertNotNull(segments, "Failed to list generated segments in " + segmentDirectory); + File newSegment = null; + for (File segment : segments) { + if (!existingNames.contains(segment.getName())) { + Assert.assertNull(newSegment, "Expected exactly one new interoperability segment"); + newSegment = segment; + } + } + Assert.assertNotNull(newSegment, "Missing generated interoperability segment"); + return newSegment; + } + + @SuppressWarnings({"rawtypes", "unchecked"}) + private static void assertSegmentPreservesVariantEnvelope(File segmentDirectory, byte[] expected) { + ImmutableSegment segment = null; + ForwardIndexReaderContext context = null; + try { + segment = ImmutableSegmentLoader.load(segmentDirectory, ReadMode.heap); + ForwardIndexReader reader = segment.getForwardIndex(PAYLOAD); + context = reader.createContext(); + byte[] actual = reader.getBytes(0, context); + Assert.assertTrue(Arrays.equals(actual, expected), + "Segment generation must preserve the unshredded producer envelope exactly"); + } catch (Exception e) { + throw new IllegalStateException("Failed to validate generated VARIANT segment", e); + } finally { + if (context != null) { + context.close(); + } + if (segment != null) { + segment.destroy(); + } + } + } + + /// Returns a valid object whose a/b/c/eventType offsets are `[12, 10, 8, 0, 14]`; its values are physically + /// encoded in eventType/c/b/a order instead of lexicographic key order. + private static byte[] nonMonotonicObjectEnvelope() { + byte[] metadata = { + 0x11, 0x04, + 0x00, 0x01, 0x02, 0x03, 0x0C, + 'a', 'b', 'c', 'e', 'v', 'e', 'n', 't', 'T', 'y', 'p', 'e' + }; + byte[] value = { + 0x02, 0x04, + 0x00, 0x01, 0x02, 0x03, + 0x0C, 0x0A, 0x08, 0x00, 0x0E, + 0x1D, 'i', 'n', 't', 'e', 'r', 'o', 'p', + 0x0C, 0x03, + 0x0C, 0x02, + 0x0C, 0x01 + }; + return VariantEnvelope.encode(metadata, 0, metadata.length, value, 0, value.length); + } + + private static InputStream openResource(String relativePath) { + String resourcePath = RESOURCE_DIRECTORY + relativePath; + InputStream inputStream = VariantTypeTest.class.getClassLoader().getResourceAsStream(resourcePath); + if (inputStream == null) { + throw new IllegalStateException("Missing VARIANT quickstart resource: " + resourcePath); + } + return inputStream; + } + + private static void assertNoExceptions(JsonNode response) { + Assert.assertEquals(response.get("exceptions").size(), 0, response.toPrettyString()); + } + + private static void assertHasException(JsonNode response) { + JsonNode exceptions = response.get("exceptions"); + Assert.assertNotNull(exceptions, response.toPrettyString()); + Assert.assertFalse(exceptions.isEmpty(), response.toPrettyString()); + } + + private static void assertExceptionContains(JsonNode response, String... expectedFragments) { + JsonNode exceptions = response.get("exceptions"); + Assert.assertNotNull(exceptions, response.toPrettyString()); + Assert.assertFalse(exceptions.isEmpty(), response.toPrettyString()); + String exceptionText = exceptions.toString().toLowerCase(Locale.ROOT); + for (String expectedFragment : expectedFragments) { + Assert.assertTrue(exceptionText.contains(expectedFragment.toLowerCase(Locale.ROOT)), + "Expected exception text to contain '" + expectedFragment + "': " + response.toPrettyString()); + } + } + + private JsonNode postVariantQuery(String query) + throws Exception { + return postQuery("SET enableNullHandling=true; " + query); + } +} diff --git a/pinot-tools/pom.xml b/pinot-tools/pom.xml index 57827e8d0a2..9ce8828beb5 100644 --- a/pinot-tools/pom.xml +++ b/pinot-tools/pom.xml @@ -327,6 +327,16 @@ </extraArguments> </jvmSettings> </program> + <program> + <mainClass>org.apache.pinot.tools.VariantQuickStart</mainClass> + <name>quick-start-variant-batch</name> + <jvmSettings> + <initialMemorySize>4G</initialMemorySize> + <extraArguments> + <extraArgument>-Dlog4j2.configurationFile=conf/quickstart-log4j2.xml</extraArgument> + </extraArguments> + </jvmSettings> + </program> <program> <mainClass>org.apache.pinot.tools.AuthQuickstart</mainClass> <name>quick-start-auth</name> diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/VariantQuickStart.java b/pinot-tools/src/main/java/org/apache/pinot/tools/VariantQuickStart.java new file mode 100644 index 00000000000..719142a2de1 --- /dev/null +++ b/pinot-tools/src/main/java/org/apache/pinot/tools/VariantQuickStart.java @@ -0,0 +1,114 @@ +/** + * 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. + */ +package org.apache.pinot.tools; + +import com.fasterxml.jackson.databind.JsonNode; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.TimeUnit; +import org.apache.pinot.tools.admin.PinotAdministrator; +import org.apache.pinot.tools.admin.command.QuickstartRunner; + + +/// Batch quickstart for ingesting and querying an Apache Parquet VARIANT column. Instances have mutable quickstart +/// lifecycle state and are intended for a single serial run, not concurrent use. +public class VariantQuickStart extends Quickstart { + private static final String[] VARIANT_TABLE_DIRECTORIES = {"examples/batch/variantEvents"}; + private static final int EXPECTED_NUM_ROWS = 5; + private static final long BOOTSTRAP_TIMEOUT_MS = TimeUnit.MINUTES.toMillis(5); + private static final long BOOTSTRAP_POLL_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1); + + @Override + public List<String> types() { + return List.of("VARIANT", "OFFLINE_VARIANT", "OFFLINE-VARIANT", "BATCH_VARIANT", "BATCH-VARIANT"); + } + + @Override + protected String[] getDefaultBatchTableDirectories() { + return VARIANT_TABLE_DIRECTORIES; + } + + @Override + protected void waitForBootstrapToComplete(QuickstartRunner runner) + throws Exception { + printStatus(Color.CYAN, + "***** Waiting up to 5 minutes for the Parquet VARIANT segment to become queryable *****"); + long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(BOOTSTRAP_TIMEOUT_MS); + JsonNode lastResponse = null; + Exception lastException = null; + while (System.nanoTime() < deadline) { + try { + lastResponse = runner.runQuery("SELECT COUNT(*) FROM variantEvents"); + JsonNode rows = lastResponse.path("resultTable").path("rows"); + if (lastResponse.path("exceptions").isEmpty() && rows.isArray() && rows.size() == 1 + && rows.get(0).isArray() && rows.get(0).size() == 1 + && rows.get(0).get(0).asInt(-1) == EXPECTED_NUM_ROWS) { + printStatus(Color.GREEN, + "***** Parquet VARIANT segment is ready with " + EXPECTED_NUM_ROWS + " rows *****"); + return; + } + } catch (Exception e) { + lastException = e; + } + Thread.sleep(BOOTSTRAP_POLL_INTERVAL_MS); + } + + String detail = lastResponse != null ? lastResponse.toString() + : lastException != null ? lastException.getMessage() : "no query response"; + throw new IllegalStateException( + "Timed out waiting for variantEvents to contain exactly " + EXPECTED_NUM_ROWS + " rows; last result: " + + detail, lastException); + } + + @Override + public void runSampleQueries(QuickstartRunner runner) + throws Exception { + printStatus(Color.YELLOW, "***** Parquet VARIANT quickstart setup complete *****"); + + runQuery(runner, "Materialized hot field", + "SELECT eventType, COUNT(*) FROM variantEvents GROUP BY eventType ORDER BY eventType"); + runQuery(runner, "Extract nested values directly from VARIANT", + "SELECT eventId, variant_get(payload, '$.user.id', 'STRING') AS userId, " + + "variant_get(payload, '$.amount', 'DOUBLE') AS amount " + + "FROM variantEvents WHERE eventType = 'checkout' ORDER BY eventId"); + runQuery(runner, "Render the retained Parquet VARIANT value as JSON", + "SELECT eventId, variantToJson(payload) FROM variantEvents ORDER BY eventId LIMIT 5"); + } + + private static void runQuery(QuickstartRunner runner, String description, String query) + throws Exception { + printStatus(Color.YELLOW, description); + printStatus(Color.CYAN, "Query : " + query); + JsonNode response = runner.runQuery("SET enableNullHandling=true; " + query); + if (!response.path("exceptions").isEmpty()) { + throw new IllegalStateException("VARIANT quickstart query failed: " + response.path("exceptions")); + } + printStatus(Color.YELLOW, prettyPrintResponse(response)); + printStatus(Color.GREEN, "***************************************************"); + } + + public static void main(String[] args) + throws Exception { + List<String> arguments = new ArrayList<>(); + arguments.addAll(Arrays.asList("QuickStart", "-type", "VARIANT")); + arguments.addAll(Arrays.asList(args)); + PinotAdministrator.main(arguments.toArray(new String[0])); + } +} diff --git a/pinot-tools/src/main/resources/examples/batch/variantEvents/README.md b/pinot-tools/src/main/resources/examples/batch/variantEvents/README.md new file mode 100644 index 00000000000..d8e8fcd0d01 --- /dev/null +++ b/pinot-tools/src/main/resources/examples/batch/variantEvents/README.md @@ -0,0 +1,88 @@ +<!-- + 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. +--> + +# Parquet VARIANT quickstart table + +`variantEvents_data.parquet` stores `payload` using the Apache Parquet +`VARIANT(1)` logical type. This sample verifies Pinot's current reader for a +top-level, non-repeated Variant column. Pinot retains the encoded value in a +raw `VARIANT` column and materializes `$.eventType` into an indexed STRING +column while ingesting the file. The ingestion spec explicitly selects +`ParquetNativeRecordReader`; use the native reader for VARIANT files that also +contain Avro schema metadata so existing Parquet reader selection remains +backward compatible. + +The persisted format, query semantics, compatibility rules, and rollout gate are +documented in the [VARIANT design document](../../../../../../../docs/design/VARIANT.md). + +Build and start the dedicated quickstart: + +```shell +./mvnw clean install -DskipTests -Pbin-dist -Pbuild-shaded-jar +build/bin/quick-start-variant-batch.sh +``` + +The quickstart creates the schema and offline table, runs the standalone +Parquet ingestion job, uploads the segment, and executes these representative +queries: + +```sql +SELECT eventType, COUNT(*) +FROM variantEvents +GROUP BY eventType +ORDER BY eventType; + +SELECT + eventId, + variant_get(payload, '$.user.id', 'STRING') AS userId, + variant_get(payload, '$.amount', 'DOUBLE') AS amount +FROM variantEvents +WHERE eventType = 'checkout' +ORDER BY eventId; + +SELECT eventId, variantToJson(payload) +FROM variantEvents +ORDER BY eventId; +``` + +The sample schema and table both enable storage null handling. VARIANT tables +must enable either schema column-based null handling or table-level null +handling so that SQL null, Variant null, and missing paths stay distinct. +All sample queries also use `SET enableNullHandling=true`; Pinot rejects +VARIANT SQL functions without that query option instead of returning ambiguous +null results. + +## Production rollout + +`VARIANT` introduces schema and query-wire type names that older Pinot +processes do not recognize. Upgrade every controller, broker, server, minion, +and external ingestion job before registering a schema containing `VARIANT`. +Keep VARIANT columns out of queries until every broker and server has been +upgraded, and do not roll back to a pre-VARIANT build while a VARIANT table is +active. Existing tables and queries are unaffected during the rolling upgrade. + +This follows the same user workflow and the supported subset of the +[Spark VARIANT function contract](https://spark.apache.org/docs/latest/api/sql/variant-functions/): +the source file carries a typed Variant value, frequently filtered paths can +be materialized during ingestion, and other paths remain available for +late-bound query-time extraction. The committed fixture was written with +parquet-java. A Spark-produced interoperability golden, nested or repeated +Variant columns, quoted path keys, streaming ingestion, and rolling +mixed-version queries over VARIANT columns are outside this sample's verified +scope. diff --git a/pinot-tools/src/main/resources/examples/batch/variantEvents/ingestionJobSpec.yaml b/pinot-tools/src/main/resources/examples/batch/variantEvents/ingestionJobSpec.yaml new file mode 100644 index 00000000000..17bad768911 --- /dev/null +++ b/pinot-tools/src/main/resources/examples/batch/variantEvents/ingestionJobSpec.yaml @@ -0,0 +1,51 @@ +# +# 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. +# + +executionFrameworkSpec: + name: 'standalone' + segmentGenerationJobRunnerClassName: 'org.apache.pinot.plugin.ingestion.batch.standalone.SegmentGenerationJobRunner' + segmentTarPushJobRunnerClassName: 'org.apache.pinot.plugin.ingestion.batch.standalone.SegmentTarPushJobRunner' + segmentUriPushJobRunnerClassName: 'org.apache.pinot.plugin.ingestion.batch.standalone.SegmentUriPushJobRunner' + segmentMetadataPushJobRunnerClassName: 'org.apache.pinot.plugin.ingestion.batch.standalone.SegmentMetadataPushJobRunner' + +jobType: SegmentCreationAndTarPush +inputDirURI: 'examples/batch/variantEvents/rawdata' +includeFileNamePattern: 'glob:**/*.parquet' +outputDirURI: 'examples/batch/variantEvents/segments' +overwriteOutput: true + +pinotFSSpecs: + - scheme: file + className: org.apache.pinot.spi.filesystem.LocalPinotFS + +recordReaderSpec: + dataFormat: 'parquet' + className: 'org.apache.pinot.plugin.inputformat.parquet.ParquetNativeRecordReader' + +tableSpec: + tableName: 'variantEvents' + schemaURI: 'http://localhost:9000/tables/variantEvents/schema' + tableConfigURI: 'http://localhost:9000/tables/variantEvents' + +pinotClusterSpecs: + - controllerURI: 'http://localhost:9000' + +pushJobSpec: + pushAttempts: 2 + pushRetryIntervalMillis: 1000 diff --git a/pinot-tools/src/main/resources/examples/batch/variantEvents/rawdata/variantEvents_data.parquet b/pinot-tools/src/main/resources/examples/batch/variantEvents/rawdata/variantEvents_data.parquet new file mode 100644 index 00000000000..0d901fde2b2 Binary files /dev/null and b/pinot-tools/src/main/resources/examples/batch/variantEvents/rawdata/variantEvents_data.parquet differ diff --git a/pinot-tools/src/main/resources/examples/batch/variantEvents/variantEvents_offline_table_config.json b/pinot-tools/src/main/resources/examples/batch/variantEvents/variantEvents_offline_table_config.json new file mode 100644 index 00000000000..99937d12be3 --- /dev/null +++ b/pinot-tools/src/main/resources/examples/batch/variantEvents/variantEvents_offline_table_config.json @@ -0,0 +1,39 @@ +{ + "tableName": "variantEvents", + "tableType": "OFFLINE", + "segmentsConfig": { + "segmentPushType": "APPEND", + "replication": "1", + "timeColumnName": "eventTime" + }, + "tenants": {}, + "fieldConfigList": [ + { + "name": "payload", + "encodingType": "RAW", + "indexes": { + "forward": { + "compressionCodec": "ZSTANDARD" + } + } + } + ], + "tableIndexConfig": { + "loadMode": "MMAP", + "nullHandlingEnabled": true, + "invertedIndexColumns": [ + "eventType" + ] + }, + "ingestionConfig": { + "transformConfigs": [ + { + "columnName": "eventType", + "transformFunction": "variant_get(payload, '$.eventType', 'STRING')" + } + ] + }, + "metadata": { + "customConfigs": {} + } +} diff --git a/pinot-tools/src/main/resources/examples/batch/variantEvents/variantEvents_schema.json b/pinot-tools/src/main/resources/examples/batch/variantEvents/variantEvents_schema.json new file mode 100644 index 00000000000..f7bce636bbe --- /dev/null +++ b/pinot-tools/src/main/resources/examples/batch/variantEvents/variantEvents_schema.json @@ -0,0 +1,26 @@ +{ + "schemaName": "variantEvents", + "enableColumnBasedNullHandling": true, + "dimensionFieldSpecs": [ + { + "name": "eventId", + "dataType": "STRING" + }, + { + "name": "eventType", + "dataType": "STRING" + }, + { + "name": "payload", + "dataType": "VARIANT" + } + ], + "dateTimeFieldSpecs": [ + { + "name": "eventTime", + "dataType": "LONG", + "format": "1:MILLISECONDS:EPOCH", + "granularity": "1:MILLISECONDS" + } + ] +} diff --git a/pinot-tools/src/test/java/org/apache/pinot/tools/admin/command/QuickStartTest.java b/pinot-tools/src/test/java/org/apache/pinot/tools/admin/command/QuickStartTest.java index fe5b386861a..1bc8c3f10bf 100644 --- a/pinot-tools/src/test/java/org/apache/pinot/tools/admin/command/QuickStartTest.java +++ b/pinot-tools/src/test/java/org/apache/pinot/tools/admin/command/QuickStartTest.java @@ -26,6 +26,7 @@ import org.apache.pinot.tools.QuickStartBase; import org.apache.pinot.tools.Quickstart; import org.apache.pinot.tools.RealtimeQuickStart; import org.apache.pinot.tools.RealtimeQuickStartWithMinion; +import org.apache.pinot.tools.VariantQuickStart; import org.testng.annotations.Test; import static org.testng.Assert.assertEquals; @@ -72,6 +73,12 @@ public class QuickStartTest { assertEquals(quickStartClassFor("MULTI_CLUSTER"), MultiClusterQuickstart.class); assertEquals(quickStartClassFor("MULTICLUSTER"), MultiClusterQuickstart.class); + + assertEquals(quickStartClassFor("VARIANT"), VariantQuickStart.class); + assertEquals(quickStartClassFor("OFFLINE_VARIANT"), VariantQuickStart.class); + assertEquals(quickStartClassFor("OFFLINE-VARIANT"), VariantQuickStart.class); + assertEquals(quickStartClassFor("BATCH_VARIANT"), VariantQuickStart.class); + assertEquals(quickStartClassFor("BATCH-VARIANT"), VariantQuickStart.class); } /// The batch-side quickstarts that only differed by their sample queries were merged into [Quickstart]; their types --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
