diff --git a/.github/workflows/maven-checks.yml b/.github/workflows/maven-checks.yml index 2295c413bef7b..a4736210597e0 100644 --- a/.github/workflows/maven-checks.yml +++ b/.github/workflows/maven-checks.yml @@ -24,7 +24,7 @@ jobs: fail-fast: false matrix: java: [ 8.0.442, 17.0.13 ] - runs-on: ubuntu-latest + runs-on: yscope-gh-runner timeout-minutes: 45 steps: - name: Free Disk Space @@ -68,7 +68,7 @@ jobs: presto-coordinator-image: name: "presto-coordinator-image" needs: "maven-checks" - runs-on: "ubuntu-22.04" + runs-on: "yscope-gh-runner" steps: - uses: "actions/checkout@v4" with: diff --git a/.github/workflows/prestissimo-worker-images-build.yml b/.github/workflows/prestissimo-worker-images-build.yml index b36dcb71949be..4ffe7c186aa95 100644 --- a/.github/workflows/prestissimo-worker-images-build.yml +++ b/.github/workflows/prestissimo-worker-images-build.yml @@ -7,12 +7,15 @@ on: jobs: prestissimo-worker-images-build: name: "prestissimo-worker-images-build" - runs-on: "ubuntu-22.04" + runs-on: "yscope-gh-runner" steps: - uses: "actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683" with: submodules: "recursive" + - name: "Set up Docker Buildx" + uses: "docker/setup-buildx-action@v3" + - name: "Login to image registry" uses: "docker/login-action@74a5d142397b4f367a81961eba4e8cd7edddf772" with: @@ -20,30 +23,62 @@ jobs: username: "${{github.actor}}" password: "${{secrets.GITHUB_TOKEN}}" + - name: "Calculate dependency hash" + id: "deps-hash" + run: |- + # Hash the dockerfile, setup scripts, and velox submodule commit + VELOX_COMMIT=$(git -C presto-native-execution/velox rev-parse HEAD) + HASH=$(cat \ + presto-native-execution/scripts/dockerfiles/ubuntu-22.04-dependency.dockerfile \ + presto-native-execution/scripts/setup-ubuntu.sh \ + presto-native-execution/scripts/setup-adapters.sh \ + presto-native-execution/velox/scripts/setup-ubuntu.sh \ + <(echo "velox:${VELOX_COMMIT}") \ + | sha256sum | cut -d' ' -f1 | cut -c1-12) + echo "hash=${HASH}" >> $GITHUB_OUTPUT + echo "Dependency hash: ${HASH}" + echo "dep_image=ghcr.io/${{github.repository}}/prestissimo-worker-dev-env:${HASH}" >> $GITHUB_OUTPUT + + - name: "Check if dependency image exists" + id: "check-deps-image" + run: |- + # Try to pull the image with the hash tag + if docker pull ghcr.io/${{github.repository}}/prestissimo-worker-dev-env:${{steps.deps-hash.outputs.hash}} 2>/dev/null; then + echo "exists=true" >> $GITHUB_OUTPUT + echo "✓ Dependency image found in cache, skipping 1+ hour build" + else + echo "exists=false" >> $GITHUB_OUTPUT + echo "✗ Dependency image not found, will build from scratch" + fi + - name: "Set up metadata for dependency image" id: "metadata-deps-image" uses: "docker/metadata-action@902fa8ec7d6ecbf8d84d538b9b233a880e428804" with: images: "ghcr.io/${{github.repository}}/prestissimo-worker-dev-env" - tags: "type=raw,value=dev" + tags: |- + type=raw,value=test + type=raw,value=${{steps.deps-hash.outputs.hash}} - name: "Build and push dependency image" + if: steps.check-deps-image.outputs.exists != 'true' uses: "docker/build-push-action@471d1dc4e07e5cdedd4c2171150001c434f0b7a4" with: context: "./presto-native-execution" file: "./presto-native-execution/scripts/dockerfiles/ubuntu-22.04-dependency.dockerfile" - push: >- - ${{github.event_name != 'pull_request' - && github.ref == 'refs/heads/release-0.293-clp-connector'}} + push: "${{github.event_name != 'pull_request'}}" tags: "${{steps.metadata-deps-image.outputs.tags}}" labels: "${{steps.metadata-deps-image.outputs.labels}}" + load: "${{github.event_name == 'pull_request'}}" + cache-from: "type=registry,ref=ghcr.io/${{github.repository}}/prestissimo-worker-dev-env:buildcache" + cache-to: "type=registry,ref=ghcr.io/${{github.repository}}/prestissimo-worker-dev-env:buildcache,mode=max" - name: "Set up metadata for runtime image" id: "metadata-runtime-image" uses: "docker/metadata-action@902fa8ec7d6ecbf8d84d538b9b233a880e428804" with: images: "ghcr.io/${{github.repository}}/prestissimo-worker" - tags: "type=raw,value=dev" + tags: "type=raw,value=test" - name: "Get number of cores" id: "get-cores" @@ -55,7 +90,7 @@ jobs: with: build-args: |- BASE_IMAGE=ubuntu:22.04 - DEPENDENCY_IMAGE=${{steps.metadata-deps-image.outputs.tags}} + DEPENDENCY_IMAGE=${{steps.deps-hash.outputs.dep_image}} EXTRA_CMAKE_FLAGS=-DPRESTO_ENABLE_TESTING=OFF \ -DPRESTO_ENABLE_PARQUET=ON \ -DPRESTO_ENABLE_S3=ON @@ -63,8 +98,6 @@ jobs: OSNAME=ubuntu context: "./presto-native-execution" file: "./presto-native-execution/scripts/dockerfiles/prestissimo-runtime.dockerfile" - push: >- - ${{github.event_name != 'pull_request' - && github.ref == 'refs/heads/release-0.293-clp-connector'}} + push: "${{github.event_name != 'pull_request'}}" tags: "${{steps.metadata-runtime-image.outputs.tags}}" labels: "${{steps.metadata-runtime-image.outputs.labels}}" diff --git a/.github/workflows/prestocpp-format-and-header-check.yml b/.github/workflows/prestocpp-format-and-header-check.yml index c554ee8785786..fe9af78d9ddc6 100644 --- a/.github/workflows/prestocpp-format-and-header-check.yml +++ b/.github/workflows/prestocpp-format-and-header-check.yml @@ -20,7 +20,7 @@ concurrency: jobs: prestocpp-format-and-header-check: - runs-on: ubuntu-latest + runs-on: yscope-gh-runner container: image: public.ecr.aws/oss-presto/velox-dev:check steps: diff --git a/.github/workflows/prestocpp-linux-build-and-unit-test.yml b/.github/workflows/prestocpp-linux-build-and-unit-test.yml index e26e330403ec2..985dfdb09ff61 100644 --- a/.github/workflows/prestocpp-linux-build-and-unit-test.yml +++ b/.github/workflows/prestocpp-linux-build-and-unit-test.yml @@ -20,7 +20,7 @@ concurrency: jobs: prestocpp-linux-build-for-test: - runs-on: ubuntu-22.04 + runs-on: yscope-gh-runner container: image: prestodb/presto-native-dependency:0.293-20250522140509-484b00e env: @@ -97,7 +97,7 @@ jobs: prestocpp-linux-presto-e2e-tests: needs: prestocpp-linux-build-for-test - runs-on: ubuntu-22.04 + runs-on: yscope-gh-runner container: image: prestodb/presto-native-dependency:0.293-20250522140509-484b00e env: @@ -172,7 +172,7 @@ jobs: prestocpp-linux-presto-native-tests: needs: prestocpp-linux-build-for-test - runs-on: ubuntu-22.04 + runs-on: yscope-gh-runner strategy: fail-fast: false matrix: @@ -248,7 +248,7 @@ jobs: prestocpp-linux-presto-sidecar-tests: needs: prestocpp-linux-build-for-test - runs-on: ubuntu-22.04 + runs-on: yscope-gh-runner container: image: prestodb/presto-native-dependency:0.293-20250522140509-484b00e env: diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 6f829502fbbb8..12c65c7b04472 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -22,7 +22,7 @@ concurrency: jobs: changes: - runs-on: ubuntu-latest + runs-on: yscope-gh-runner # Required permissions permissions: pull-requests: read @@ -41,7 +41,7 @@ jobs: - '!presto-docs/**' test: - runs-on: ubuntu-latest + runs-on: yscope-gh-runner needs: changes strategy: fail-fast: false diff --git a/presto-clp/pom.xml b/presto-clp/pom.xml index d13b59968aff5..8941da517cecb 100644 --- a/presto-clp/pom.xml +++ b/presto-clp/pom.xml @@ -75,6 +75,11 @@ jackson-databind + + com.fasterxml.jackson.dataformat + jackson-dataformat-yaml + + com.facebook.presto presto-spi @@ -154,4 +159,14 @@ test + + + + + org.yaml + snakeyaml + 2.1 + + + diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpConfig.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpConfig.java index 121eb0d5ff17b..329d5451c71a5 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpConfig.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpConfig.java @@ -33,6 +33,8 @@ public class ClpConfig private long metadataRefreshInterval = 60; private long metadataExpireInterval = 600; + private String metadataYamlPath; + private String splitFilterConfig; private SplitFilterProviderType splitFilterProviderType = SplitFilterProviderType.MYSQL; private SplitProviderType splitProviderType = SplitProviderType.MYSQL; @@ -151,6 +153,18 @@ public ClpConfig setMetadataExpireInterval(long metadataExpireInterval) return this; } + public String getMetadataYamlPath() + { + return metadataYamlPath; + } + + @Config("clp.metadata-yaml-path") + public ClpConfig setMetadataYamlPath(String metadataYamlPath) + { + this.metadataYamlPath = metadataYamlPath; + return this; + } + public String getSplitFilterConfig() { return splitFilterConfig; @@ -189,7 +203,8 @@ public ClpConfig setSplitProviderType(SplitProviderType splitProviderType) public enum MetadataProviderType { - MYSQL + MYSQL, + YAML } public enum SplitFilterProviderType @@ -199,6 +214,7 @@ public enum SplitFilterProviderType public enum SplitProviderType { - MYSQL + MYSQL, + PINOT } } diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpErrorCode.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpErrorCode.java index 2530c013455cc..fb6626de25a61 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpErrorCode.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpErrorCode.java @@ -28,6 +28,7 @@ public enum ClpErrorCode CLP_UNSUPPORTED_SPLIT_SOURCE(2, EXTERNAL), CLP_UNSUPPORTED_TYPE(3, EXTERNAL), CLP_UNSUPPORTED_CONFIG_OPTION(4, EXTERNAL), + CLP_UNSUPPORTED_TABLE_SCHEMA_YAML(5, EXTERNAL), CLP_SPLIT_FILTER_CONFIG_NOT_FOUND(10, USER_ERROR), CLP_MANDATORY_SPLIT_FILTER_NOT_VALID(11, USER_ERROR), diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpModule.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpModule.java index bf801d0d87242..6d979ca0bceb8 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpModule.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpModule.java @@ -16,7 +16,9 @@ import com.facebook.airlift.configuration.AbstractConfigurationAwareModule; import com.facebook.presto.plugin.clp.metadata.ClpMetadataProvider; import com.facebook.presto.plugin.clp.metadata.ClpMySqlMetadataProvider; +import com.facebook.presto.plugin.clp.metadata.ClpYamlMetadataProvider; import com.facebook.presto.plugin.clp.split.ClpMySqlSplitProvider; +import com.facebook.presto.plugin.clp.split.ClpPinotSplitProvider; import com.facebook.presto.plugin.clp.split.ClpSplitProvider; import com.facebook.presto.plugin.clp.split.filter.ClpMySqlSplitFilterProvider; import com.facebook.presto.plugin.clp.split.filter.ClpSplitFilterProvider; @@ -56,6 +58,9 @@ protected void setup(Binder binder) if (config.getMetadataProviderType() == MetadataProviderType.MYSQL) { binder.bind(ClpMetadataProvider.class).to(ClpMySqlMetadataProvider.class).in(Scopes.SINGLETON); } + else if (config.getMetadataProviderType() == MetadataProviderType.YAML) { + binder.bind(ClpMetadataProvider.class).to(ClpYamlMetadataProvider.class).in(Scopes.SINGLETON); + } else { throw new PrestoException(CLP_UNSUPPORTED_METADATA_SOURCE, "Unsupported metadata provider type: " + config.getMetadataProviderType()); } @@ -63,6 +68,9 @@ protected void setup(Binder binder) if (config.getSplitProviderType() == SplitProviderType.MYSQL) { binder.bind(ClpSplitProvider.class).to(ClpMySqlSplitProvider.class).in(Scopes.SINGLETON); } + else if (config.getSplitProviderType() == SplitProviderType.PINOT) { + binder.bind(ClpSplitProvider.class).to(ClpPinotSplitProvider.class).in(Scopes.SINGLETON); + } else { throw new PrestoException(CLP_UNSUPPORTED_SPLIT_SOURCE, "Unsupported split provider type: " + config.getSplitProviderType()); } diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpSplit.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpSplit.java index 2e35840971c11..7b2d42bb0635d 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpSplit.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpSplit.java @@ -24,6 +24,7 @@ import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import static com.facebook.presto.spi.schedule.NodeSelectionStrategy.NO_PREFERENCE; @@ -77,6 +78,25 @@ public List getPreferredNodes(NodeProvider nodeProvider) return ImmutableList.of(); } + @Override + public int hashCode() + { + return Objects.hash(path, type, kqlQuery); + } + + @Override + public boolean equals(Object obj) + { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + ClpSplit other = (ClpSplit) obj; + return this.type == other.type && this.path.equals(other.path) && this.kqlQuery.equals(other.kqlQuery); + } + @Override public Map getInfo() { diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpTableLayoutHandle.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpTableLayoutHandle.java index b82932f0c30fd..902c9dfe37176 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpTableLayoutHandle.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpTableLayoutHandle.java @@ -13,6 +13,7 @@ */ package com.facebook.presto.plugin.clp; +import com.facebook.presto.plugin.clp.optimization.ClpTopNSpec; import com.facebook.presto.spi.ConnectorTableLayoutHandle; import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; @@ -28,13 +29,34 @@ public class ClpTableLayoutHandle private final ClpTableHandle table; private final Optional kqlQuery; private final Optional metadataSql; + private final boolean metadataQueryOnly; + private final Optional topN; @JsonCreator - public ClpTableLayoutHandle(@JsonProperty("table") ClpTableHandle table, @JsonProperty("kqlQuery") Optional kqlQuery, @JsonProperty("metadataFilterQuery") Optional metadataSql) + public ClpTableLayoutHandle( + @JsonProperty("table") ClpTableHandle table, + @JsonProperty("kqlQuery") Optional kqlQuery, + @JsonProperty("metadataFilterQuery") Optional metadataSql, + @JsonProperty("metadataQueryOnly") boolean metadataQueryOnly, + @JsonProperty("topN") Optional topN) { this.table = table; this.kqlQuery = kqlQuery; this.metadataSql = metadataSql; + this.metadataQueryOnly = metadataQueryOnly; + this.topN = topN; + } + + public ClpTableLayoutHandle( + @JsonProperty("table") ClpTableHandle table, + @JsonProperty("kqlQuery") Optional kqlQuery, + @JsonProperty("metadataFilterQuery") Optional metadataSql) + { + this.table = table; + this.kqlQuery = kqlQuery; + this.metadataSql = metadataSql; + this.metadataQueryOnly = false; + this.topN = Optional.empty(); } @JsonProperty @@ -55,6 +77,18 @@ public Optional getMetadataSql() return metadataSql; } + @JsonProperty + public boolean isMetadataQueryOnly() + { + return metadataQueryOnly; + } + + @JsonProperty + public Optional getTopN() + { + return topN; + } + @Override public boolean equals(Object o) { @@ -67,13 +101,15 @@ public boolean equals(Object o) ClpTableLayoutHandle that = (ClpTableLayoutHandle) o; return Objects.equals(table, that.table) && Objects.equals(kqlQuery, that.kqlQuery) && - Objects.equals(metadataSql, that.metadataSql); + Objects.equals(metadataSql, that.metadataSql) && + Objects.equals(metadataQueryOnly, that.metadataQueryOnly) && + Objects.equals(topN, that.topN); } @Override public int hashCode() { - return Objects.hash(table, kqlQuery, metadataSql); + return Objects.hash(table, kqlQuery, metadataSql, metadataQueryOnly, topN); } @Override @@ -83,6 +119,8 @@ public String toString() .add("table", table) .add("kqlQuery", kqlQuery) .add("metadataSql", metadataSql) + .add("metadataQueryOnly", metadataQueryOnly) + .add("topN", topN) .toString(); } } diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/metadata/ClpYamlMetadataProvider.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/metadata/ClpYamlMetadataProvider.java new file mode 100644 index 0000000000000..0bbb0b4c9ea79 --- /dev/null +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/metadata/ClpYamlMetadataProvider.java @@ -0,0 +1,140 @@ +/* + * Licensed 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 com.facebook.presto.plugin.clp.metadata; + +import com.facebook.airlift.log.Logger; +import com.facebook.presto.plugin.clp.ClpColumnHandle; +import com.facebook.presto.plugin.clp.ClpConfig; +import com.facebook.presto.plugin.clp.ClpTableHandle; +import com.facebook.presto.spi.PrestoException; +import com.facebook.presto.spi.SchemaTableName; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.dataformat.yaml.YAMLFactory; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; + +import javax.inject.Inject; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static com.facebook.presto.plugin.clp.ClpConnectorFactory.CONNECTOR_NAME; +import static com.facebook.presto.plugin.clp.ClpErrorCode.CLP_UNSUPPORTED_TABLE_SCHEMA_YAML; +import static java.lang.String.format; + +public class ClpYamlMetadataProvider + implements ClpMetadataProvider +{ + private static final Logger log = Logger.get(ClpYamlMetadataProvider.class); + private final ClpConfig config; + private Map tableSchemaYamlMap; + + @Inject + public ClpYamlMetadataProvider(ClpConfig config) + { + this.config = config; + } + + @Override + public List listColumnHandles(SchemaTableName schemaTableName) + { + Path tableSchemaPath = Paths.get(tableSchemaYamlMap.get(schemaTableName)); + ClpSchemaTree schemaTree = new ClpSchemaTree(config.isPolymorphicTypeEnabled()); + ObjectMapper mapper = new ObjectMapper(new YAMLFactory()); + + try { + Map root = mapper.readValue( + new File(tableSchemaPath.toString()), + new TypeReference>() {}); + ImmutableList.Builder namesBuilder = ImmutableList.builder(); + ImmutableList.Builder typesBuilder = ImmutableList.builder(); + collectTypes(root, "", namesBuilder, typesBuilder); + ImmutableList names = namesBuilder.build(); + ImmutableList types = typesBuilder.build(); + // The names and types should have same sizes + for (int i = 0; i < names.size(); i++) { + schemaTree.addColumn(names.get(i), types.get(i)); + } + return schemaTree.collectColumnHandles(); + } + catch (IOException e) { + log.error(format("Failed to parse table schema file %s, error: %s", tableSchemaPath, e.getMessage()), e); + } + return Collections.emptyList(); + } + + @Override + public List listTableHandles(String schemaName) + { + Path tablesSchemaPath = Paths.get(config.getMetadataYamlPath()); + ObjectMapper mapper = new ObjectMapper(new YAMLFactory()); + + try { + Map root = mapper.readValue(new File(tablesSchemaPath.toString()), + new TypeReference>() {}); + + Object catalogObj = root.get(CONNECTOR_NAME); + if (!(catalogObj instanceof Map)) { + throw new PrestoException(CLP_UNSUPPORTED_TABLE_SCHEMA_YAML, format("The table schema does not contain field: %s", CONNECTOR_NAME)); + } + Object schemaObj = ((Map) catalogObj).get(schemaName); + ImmutableList.Builder tableHandlesBuilder = new ImmutableList.Builder<>(); + ImmutableMap.Builder tableSchemaYamlMapBuilder = new ImmutableMap.Builder<>(); + for (Map.Entry schemaEntry : ((Map) schemaObj).entrySet()) { + String tableName = schemaEntry.getKey(); + String tableSchemaYamlPath = schemaEntry.getValue().toString(); + // The splits' absolute paths will be stored in Pinot metadata database + SchemaTableName schemaTableName = new SchemaTableName(schemaName, tableName); + tableHandlesBuilder.add(new ClpTableHandle(schemaTableName, "")); + tableSchemaYamlMapBuilder.put(schemaTableName, tableSchemaYamlPath); + } + this.tableSchemaYamlMap = tableSchemaYamlMapBuilder.build(); + return tableHandlesBuilder.build(); + } + catch (IOException e) { + log.error(format("Failed to parse metadata file: %s, error: %s", config.getMetadataYamlPath(), e.getMessage()), e); + } + return Collections.emptyList(); + } + + private void collectTypes(Object node, String prefix, ImmutableList.Builder namesBuilder, ImmutableList.Builder typesBuilder) + { + if (node instanceof Number) { + namesBuilder.add(prefix); + typesBuilder.add(((Number) node).byteValue()); + return; + } + if (node instanceof List) { + for (Number type : (List) node) { + namesBuilder.add(prefix); + typesBuilder.add(type.byteValue()); + } + return; + } + for (Map.Entry entry : ((Map) node).entrySet()) { + if (!prefix.isEmpty()) { + collectTypes(entry.getValue(), format("%s.%s", prefix, entry.getKey()), namesBuilder, typesBuilder); + continue; + } + collectTypes(entry.getValue(), entry.getKey(), namesBuilder, typesBuilder); + } + } +} diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpComputePushDown.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpComputePushDown.java index 2c216614af10f..c86ea0dbe4d7e 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpComputePushDown.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpComputePushDown.java @@ -14,7 +14,10 @@ package com.facebook.presto.plugin.clp.optimization; import com.facebook.airlift.log.Logger; -import com.facebook.presto.plugin.clp.ClpExpression; +import com.facebook.presto.common.block.SortOrder; +import com.facebook.presto.common.type.RowType; +import com.facebook.presto.plugin.clp.ClpColumnHandle; +import com.facebook.presto.plugin.clp.ClpMetadata; import com.facebook.presto.plugin.clp.ClpTableHandle; import com.facebook.presto.plugin.clp.ClpTableLayoutHandle; import com.facebook.presto.plugin.clp.split.filter.ClpSplitFilterProvider; @@ -22,25 +25,38 @@ import com.facebook.presto.spi.ConnectorPlanOptimizer; import com.facebook.presto.spi.ConnectorPlanRewriter; import com.facebook.presto.spi.ConnectorSession; +import com.facebook.presto.spi.ConnectorTableLayoutHandle; import com.facebook.presto.spi.TableHandle; import com.facebook.presto.spi.VariableAllocator; import com.facebook.presto.spi.function.FunctionMetadataManager; import com.facebook.presto.spi.function.StandardFunctionResolution; import com.facebook.presto.spi.plan.FilterNode; +import com.facebook.presto.spi.plan.Ordering; import com.facebook.presto.spi.plan.PlanNode; import com.facebook.presto.spi.plan.PlanNodeIdAllocator; +import com.facebook.presto.spi.plan.ProjectNode; import com.facebook.presto.spi.plan.TableScanNode; +import com.facebook.presto.spi.plan.TopNNode; +import com.facebook.presto.spi.relation.ConstantExpression; import com.facebook.presto.spi.relation.RowExpression; +import com.facebook.presto.spi.relation.SpecialFormExpression; import com.facebook.presto.spi.relation.VariableReferenceExpression; +import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableSet; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; import java.util.HashSet; +import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.Set; import static com.facebook.presto.plugin.clp.ClpConnectorFactory.CONNECTOR_NAME; import static com.facebook.presto.spi.ConnectorPlanRewriter.rewriteWith; +import static java.lang.Math.toIntExact; import static java.lang.String.format; import static java.util.Objects.requireNonNull; @@ -67,7 +83,7 @@ public PlanNode optimize(PlanNode maxSubplan, ConnectorSession session, Variable // Throw exception if any required split filters are missing if (!rewriter.tableScopeSet.isEmpty() && !rewriter.hasVisitedFilter) { - splitFilterProvider.checkContainsRequiredFilters(rewriter.tableScopeSet, ""); + splitFilterProvider.checkContainsRequiredFilters(rewriter.tableScopeSet, ImmutableSet.of()); } return optimizedPlanNode; } @@ -105,6 +121,156 @@ public PlanNode visitFilter(FilterNode node, RewriteContext context) return processFilter(node, (TableScanNode) node.getSource()); } + @Override + public PlanNode visitTopN(TopNNode node, RewriteContext context) + { + PlanNode rewrittenSource = context.rewrite(node.getSource(), null); + + ProjectNode project = null; + FilterNode filter = null; + PlanNode cursor = rewrittenSource; + + if (cursor instanceof ProjectNode) { + project = (ProjectNode) cursor; + cursor = project.getSource(); + } + if (cursor instanceof FilterNode) { + filter = (FilterNode) cursor; + cursor = filter.getSource(); + } + if (!(cursor instanceof TableScanNode)) { + return node.replaceChildren(ImmutableList.of(rewrittenSource)); + } + + TableScanNode scan = (TableScanNode) cursor; + TableHandle tableHandle = scan.getTable(); + if (!(tableHandle.getConnectorHandle() instanceof ClpTableHandle)) { + return node.replaceChildren(ImmutableList.of(rewrittenSource)); + } + + // only allow TopN pushdown when metadata-only is true + boolean metadataOnly = false; + Optional layout = tableHandle.getLayout(); + Optional kql = Optional.empty(); + Optional metadataSql = Optional.empty(); + Optional existingTopN = Optional.empty(); + ClpTableHandle clpTableHandle = null; + + if (layout.isPresent() && layout.get() instanceof ClpTableLayoutHandle) { + ClpTableLayoutHandle cl = (ClpTableLayoutHandle) layout.get(); + metadataOnly = cl.isMetadataQueryOnly(); + kql = cl.getKqlQuery(); + metadataSql = cl.getMetadataSql(); + existingTopN = cl.getTopN(); + clpTableHandle = cl.getTable(); + } + + if (!metadataOnly) { + // Rule: skip TopN pushdown unless metadataQueryOnly is true + return node.replaceChildren(ImmutableList.of(rewrittenSource)); + } + + // Ensure ORDER BY items are plain variables (allow identity through Project) + List ords = node.getOrderingScheme().getOrderBy(); + if (project != null && !areIdents(project, ords)) { + return node.replaceChildren(ImmutableList.of(rewrittenSource)); + } + + Map assignments = scan.getAssignments(); + List newOrderings = new ArrayList<>(ords.size()); + for (Ordering ord : ords) { + VariableReferenceExpression outVar = ord.getVariable(); + Optional columnNameOpt = buildOrderColumnName(project, outVar, assignments); + if (!columnNameOpt.isPresent()) { + return node.replaceChildren(ImmutableList.of(rewrittenSource)); + } + + String tableScope = CONNECTOR_NAME + "." + (clpTableHandle != null ? + clpTableHandle.getSchemaTableName().toString() : ClpMetadata.DEFAULT_SCHEMA_NAME); + + List remappedColumnName = splitFilterProvider.remapColumnName(tableScope, columnNameOpt.get()); + newOrderings.add(new ClpTopNSpec.Ordering(remappedColumnName, toClpOrder(ord.getSortOrder()))); + } + + if (existingTopN.isPresent()) { + ClpTopNSpec ex = existingTopN.get(); + if (!sameOrdering(ex.getOrderings(), newOrderings)) { + return node.replaceChildren(ImmutableList.of(rewrittenSource)); // leave existing as-is + } + long mergedLimit = Math.min(ex.getLimit(), node.getCount()); + if (mergedLimit == ex.getLimit()) { + // No change needed; keep current layout/spec + return node.replaceChildren(ImmutableList.of(rewrittenSource)); + } + + // Tighten the limit on the layout + ClpTopNSpec tightened = new ClpTopNSpec(mergedLimit, ex.getOrderings()); + ClpTableHandle clpHandle = (ClpTableHandle) tableHandle.getConnectorHandle(); + ClpTableLayoutHandle newLayout = + new ClpTableLayoutHandle(clpHandle, kql, metadataSql, true, Optional.of(tightened)); + + TableScanNode newScan = new TableScanNode( + scan.getSourceLocation(), + idAllocator.getNextId(), + new TableHandle( + tableHandle.getConnectorId(), + clpHandle, + tableHandle.getTransaction(), + Optional.of(newLayout)), + scan.getOutputVariables(), + scan.getAssignments(), + scan.getTableConstraints(), + scan.getCurrentConstraint(), + scan.getEnforcedConstraint(), + scan.getCteMaterializationInfo()); + + PlanNode newSource = newScan; + if (filter != null) { + newSource = new FilterNode(filter.getSourceLocation(), idAllocator.getNextId(), newSource, filter.getPredicate()); + } + if (project != null) { + newSource = new ProjectNode( + project.getSourceLocation(), + idAllocator.getNextId(), + newSource, + project.getAssignments(), + project.getLocality()); + } + + return new TopNNode(node.getSourceLocation(), idAllocator.getNextId(), newSource, node.getCount(), node.getOrderingScheme(), node.getStep()); + } + + ClpTopNSpec spec = new ClpTopNSpec(node.getCount(), newOrderings); + ClpTableHandle clpHandle = (ClpTableHandle) tableHandle.getConnectorHandle(); + ClpTableLayoutHandle newLayout = + new ClpTableLayoutHandle(clpHandle, kql, metadataSql, true, Optional.of(spec)); + + TableScanNode newScanNode = new TableScanNode( + scan.getSourceLocation(), + idAllocator.getNextId(), + new TableHandle( + tableHandle.getConnectorId(), + clpHandle, + tableHandle.getTransaction(), + Optional.of(newLayout)), + scan.getOutputVariables(), + scan.getAssignments(), + scan.getTableConstraints(), + scan.getCurrentConstraint(), + scan.getEnforcedConstraint(), + scan.getCteMaterializationInfo()); + + PlanNode newSource = newScanNode; + if (filter != null) { + newSource = new FilterNode(filter.getSourceLocation(), idAllocator.getNextId(), newSource, filter.getPredicate()); + } + if (project != null) { + newSource = new ProjectNode(project.getSourceLocation(), idAllocator.getNextId(), newSource, project.getAssignments(), project.getLocality()); + } + + return new TopNNode(node.getSourceLocation(), idAllocator.getNextId(), newSource, node.getCount(), node.getOrderingScheme(), node.getStep()); + } + private PlanNode processFilter(FilterNode filterNode, TableScanNode tableScanNode) { hasVisitedFilter = true; @@ -114,21 +280,23 @@ private PlanNode processFilter(FilterNode filterNode, TableScanNode tableScanNod String tableScope = CONNECTOR_NAME + "." + clpTableHandle.getSchemaTableName().toString(); Map assignments = tableScanNode.getAssignments(); + Set metadataColumnNames = splitFilterProvider.getColumnNames(tableScope); ClpExpression clpExpression = filterNode.getPredicate().accept( new ClpFilterToKqlConverter( functionResolution, functionManager, assignments, - splitFilterProvider.getColumnNames(tableScope)), + metadataColumnNames), null); + Optional kqlQuery = clpExpression.getPushDownExpression(); Optional metadataSqlQuery = clpExpression.getMetadataSqlQuery(); Optional remainingPredicate = clpExpression.getRemainingExpression(); // Perform required metadata filter checks before handling the KQL query (if kqlQuery // isn't present, we'll return early, skipping subsequent checks). - splitFilterProvider.checkContainsRequiredFilters(ImmutableSet.of(tableScope), metadataSqlQuery.orElse("")); + splitFilterProvider.checkContainsRequiredFilters(ImmutableSet.of(tableScope), clpExpression.getPushDownVariables()); boolean hasMetadataFilter = metadataSqlQuery.isPresent() && !metadataSqlQuery.get().isEmpty(); if (hasMetadataFilter) { metadataSqlQuery = Optional.of(splitFilterProvider.remapSplitFilterPushDownExpression(tableScope, metadataSqlQuery.get())); @@ -140,7 +308,12 @@ private PlanNode processFilter(FilterNode filterNode, TableScanNode tableScanNod log.debug("KQL query: %s", kqlQuery.get()); } - ClpTableLayoutHandle layoutHandle = new ClpTableLayoutHandle(clpTableHandle, kqlQuery, metadataSqlQuery); + ClpTableLayoutHandle layoutHandle = new ClpTableLayoutHandle( + clpTableHandle, + kqlQuery, + metadataSqlQuery, + metadataColumnNames.equals(clpExpression.getPushDownVariables()), + Optional.empty()); TableHandle newTableHandle = new TableHandle( tableHandle.getConnectorId(), clpTableHandle, @@ -171,5 +344,141 @@ private PlanNode processFilter(FilterNode filterNode, TableScanNode tableScanNod return tableScanNode; } } + + private boolean sameOrdering(List a, List b) + { + if (a.size() != b.size()) { + return false; + } + for (int i = 0; i < a.size(); i++) { + ClpTopNSpec.Ordering x = a.get(i); + ClpTopNSpec.Ordering y = b.get(i); + if (!Objects.equals(x.getColumns(), y.getColumns())) { + return false; + } + if (x.getOrder() != y.getOrder()) { + return false; + } + } + return true; + } + + /** Accept plain var or dereference-of-var passthroughs. */ + private boolean areIdents(ProjectNode project, List vars) + { + for (Ordering ord : vars) { + VariableReferenceExpression out = ord.getVariable(); + RowExpression expr = project.getAssignments().get(out); + + if (expr instanceof VariableReferenceExpression) { + continue; + } + if (isDereferenceChainOverVariable(expr)) { + continue; + } + return false; + } + return true; + } + + /** Build final column name string for CLP (e.g., "msg.timestamp"), or empty if not pushdownable. */ + private Optional buildOrderColumnName( + ProjectNode project, + VariableReferenceExpression outVar, + Map assignments) + { + if (project == null) { + // ORDER BY directly on scan var + ColumnHandle ch = assignments.get(outVar); + if (!(ch instanceof ClpColumnHandle)) { + return Optional.empty(); + } + return Optional.of(((ClpColumnHandle) ch).getOriginalColumnName()); + } + + RowExpression expr = project.getAssignments().get(outVar); + if (expr instanceof VariableReferenceExpression) { + ColumnHandle ch = assignments.get((VariableReferenceExpression) expr); + if (!(ch instanceof ClpColumnHandle)) { + return Optional.empty(); + } + return Optional.of(((ClpColumnHandle) ch).getOriginalColumnName()); + } + + // Handle DEREFERENCE chain: baseVar.field1.field2... + Deque path = new ArrayDeque<>(); + RowExpression cur = expr; + + while (cur instanceof SpecialFormExpression + && ((SpecialFormExpression) cur).getForm() == SpecialFormExpression.Form.DEREFERENCE) { + SpecialFormExpression s = (SpecialFormExpression) cur; + RowExpression base = s.getArguments().get(0); + RowExpression indexExpr = s.getArguments().get(1); + + if (!(indexExpr instanceof ConstantExpression) || !(base.getType() instanceof RowType)) { + return Optional.empty(); + } + int idx; + Object v = ((ConstantExpression) indexExpr).getValue(); + if (v instanceof Long) { + idx = toIntExact((Long) v); + } + else if (v instanceof Integer) { + idx = (Integer) v; + } + else { + return Optional.empty(); + } + + RowType rowType = (RowType) base.getType(); + if (idx < 0 || idx >= rowType.getFields().size()) { + return Optional.empty(); + } + String fname = rowType.getFields().get(idx).getName().orElse(String.valueOf(idx)); + // We traverse outer->inner; collect in deque and join later + path.addLast(fname); + + cur = base; // move up the chain + } + + if (!(cur instanceof VariableReferenceExpression)) { + return Optional.empty(); + } + + ColumnHandle baseCh = assignments.get((VariableReferenceExpression) cur); + if (!(baseCh instanceof ClpColumnHandle)) { + return Optional.empty(); + } + + String baseName = ((ClpColumnHandle) baseCh).getOriginalColumnName(); + if (path.isEmpty()) { + return Optional.of(baseName); + } + return Optional.of(baseName + "." + String.join(".", path)); + } + + /** True if expr is DEREFERENCE(... DEREFERENCE(baseVar, i) ..., j) with baseVar a VariableReferenceExpression. */ + private boolean isDereferenceChainOverVariable(RowExpression expr) + { + RowExpression cur = expr; + while (cur instanceof SpecialFormExpression + && ((SpecialFormExpression) cur).getForm() == SpecialFormExpression.Form.DEREFERENCE) { + cur = ((SpecialFormExpression) cur).getArguments().get(0); + } + return (cur instanceof VariableReferenceExpression); + } + + private ClpTopNSpec.Order toClpOrder(SortOrder so) + { + switch (so) { + case ASC_NULLS_FIRST: + case ASC_NULLS_LAST: + return ClpTopNSpec.Order.ASC; + case DESC_NULLS_FIRST: + case DESC_NULLS_LAST: + return ClpTopNSpec.Order.DESC; + default: throw new IllegalArgumentException("Unknown sort order: " + so); + } + } } } diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpExpression.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpExpression.java similarity index 65% rename from presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpExpression.java rename to presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpExpression.java index e970f9848a9cf..571ecb028dc0a 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/ClpExpression.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpExpression.java @@ -11,11 +11,13 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package com.facebook.presto.plugin.clp; +package com.facebook.presto.plugin.clp.optimization; import com.facebook.presto.spi.relation.RowExpression; +import com.google.common.collect.ImmutableSet; import java.util.Optional; +import java.util.Set; /** * Represents the result of: @@ -38,11 +40,19 @@ public class ClpExpression // The remaining (non-translatable) portion of the RowExpression, if any. private final Optional remainingExpression; - public ClpExpression(String pushDownExpression, String metadataSqlQuery, RowExpression remainingExpression) + // Variables used in pushDownExpression + private final Set pushDownVariables; + + public ClpExpression( + String pushDownExpression, + String metadataSqlQuery, + RowExpression remainingExpression, + Set pushDownVariables) { this.pushDownExpression = Optional.ofNullable(pushDownExpression); this.metadataSqlQuery = Optional.ofNullable(metadataSqlQuery); this.remainingExpression = Optional.ofNullable(remainingExpression); + this.pushDownVariables = ImmutableSet.copyOf(pushDownVariables); } /** @@ -50,7 +60,7 @@ public ClpExpression(String pushDownExpression, String metadataSqlQuery, RowExpr */ public ClpExpression() { - this(null, null, null); + this(null, null, null, ImmutableSet.of()); } /** @@ -60,7 +70,18 @@ public ClpExpression() */ public ClpExpression(String pushDownExpression) { - this(pushDownExpression, null, null); + this(pushDownExpression, null, null, ImmutableSet.of()); + } + + /** + * Creates a ClpExpression from a fully translatable KQL query or column name. + * + * @param pushDownExpression + * @param pushDownVariables + */ + public ClpExpression(String pushDownExpression, Set pushDownVariables) + { + this(pushDownExpression, null, null, pushDownVariables); } /** @@ -72,7 +93,20 @@ public ClpExpression(String pushDownExpression) */ public ClpExpression(String pushDownExpression, String metadataSqlQuery) { - this(pushDownExpression, metadataSqlQuery, null); + this(pushDownExpression, metadataSqlQuery, null, ImmutableSet.of()); + } + + /** + * Creates a ClpExpression from a fully translatable KQL string or column name, as well as a + * metadata SQL string. + * + * @param pushDownExpression + * @param metadataSqlQuery + * @param pushDownVariables + */ + public ClpExpression(String pushDownExpression, String metadataSqlQuery, Set pushDownVariables) + { + this(pushDownExpression, metadataSqlQuery, null, pushDownVariables); } /** @@ -82,7 +116,7 @@ public ClpExpression(String pushDownExpression, String metadataSqlQuery) */ public ClpExpression(RowExpression remainingExpression) { - this(null, null, remainingExpression); + this(null, null, remainingExpression, ImmutableSet.of()); } public Optional getPushDownExpression() @@ -99,4 +133,9 @@ public Optional getRemainingExpression() { return remainingExpression; } + + public Set getPushDownVariables() + { + return pushDownVariables; + } } diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpFilterToKqlConverter.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpFilterToKqlConverter.java index b27a61ef0d65a..020f935163dde 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpFilterToKqlConverter.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpFilterToKqlConverter.java @@ -16,10 +16,10 @@ import com.facebook.presto.common.function.OperatorType; import com.facebook.presto.common.type.DecimalType; import com.facebook.presto.common.type.RowType; +import com.facebook.presto.common.type.TimestampType; import com.facebook.presto.common.type.Type; import com.facebook.presto.common.type.VarcharType; import com.facebook.presto.plugin.clp.ClpColumnHandle; -import com.facebook.presto.plugin.clp.ClpExpression; import com.facebook.presto.spi.ColumnHandle; import com.facebook.presto.spi.PrestoException; import com.facebook.presto.spi.function.FunctionHandle; @@ -65,6 +65,7 @@ import static java.lang.Integer.parseInt; import static java.lang.String.format; import static java.util.Objects.requireNonNull; +import static java.util.concurrent.TimeUnit.SECONDS; /** * A translator to translate Presto {@link RowExpression}s into: @@ -163,7 +164,8 @@ public ClpExpression visitConstant(ConstantExpression node, Void context) @Override public ClpExpression visitVariableReference(VariableReferenceExpression node, Void context) { - return new ClpExpression(getVariableName(node)); + String variableName = getVariableName(node); + return new ClpExpression(variableName, ImmutableSet.of(variableName)); } @Override @@ -250,7 +252,8 @@ private ClpExpression handleBetween(CallExpression node) return new ClpExpression(node); } - Optional variableOpt = first.accept(this, null).getPushDownExpression(); + ClpExpression variableExpression = first.accept(this, null); + Optional variableOpt = variableExpression.getPushDownExpression(); if (!variableOpt.isPresent() || !(second instanceof ConstantExpression) || !(third instanceof ConstantExpression)) { @@ -258,13 +261,15 @@ private ClpExpression handleBetween(CallExpression node) } String variable = variableOpt.get(); - String lowerBound = getLiteralString((ConstantExpression) second); - String upperBound = getLiteralString((ConstantExpression) third); + Type lowerBoundType = second.getType(); + String lowerBound = tryEnsureNanosecondTimestamp(lowerBoundType, getLiteralString((ConstantExpression) second)); + Type upperBoundType = third.getType(); + String upperBound = tryEnsureNanosecondTimestamp(upperBoundType, getLiteralString((ConstantExpression) third)); String kql = String.format("%s >= %s AND %s <= %s", variable, lowerBound, variable, upperBound); String metadataSqlQuery = metadataFilterColumns.contains(variable) ? String.format("\"%s\" >= %s AND \"%s\" <= %s", variable, lowerBound, variable, upperBound) : null; - return new ClpExpression(kql, metadataSqlQuery); + return new ClpExpression(kql, metadataSqlQuery, variableExpression.getPushDownVariables()); } /** @@ -290,10 +295,10 @@ private ClpExpression handleNot(CallExpression node) } String notPushDownExpression = "NOT " + expression.getPushDownExpression().get(); if (expression.getMetadataSqlQuery().isPresent()) { - return new ClpExpression(notPushDownExpression, "NOT " + expression.getMetadataSqlQuery()); + return new ClpExpression(notPushDownExpression, "NOT " + expression.getMetadataSqlQuery(), expression.getPushDownVariables()); } else { - return new ClpExpression(notPushDownExpression); + return new ClpExpression(notPushDownExpression, expression.getPushDownVariables()); } } @@ -345,7 +350,7 @@ else if (argument instanceof CallExpression) { return new ClpExpression(node); } pattern = pattern.replace("%", "*").replace("_", "?"); - return new ClpExpression(format("%s: \"%s\"", variableName, pattern)); + return new ClpExpression(format("%s: \"%s\"", variableName, pattern), variable.getPushDownVariables()); } /** @@ -442,33 +447,39 @@ private ClpExpression buildClpExpression( RowExpression originalNode) { String metadataSqlQuery = null; + literalString = tryEnsureNanosecondTimestamp(literalType, literalString); if (operator.equals(EQUAL)) { if (literalType instanceof VarcharType) { - return new ClpExpression(format("%s: \"%s\"", variableName, escapeKqlSpecialCharsForStringValue(literalString))); + return new ClpExpression( + format("%s: \"%s\"", variableName, escapeKqlSpecialCharsForStringValue(literalString)), + ImmutableSet.of(variableName)); } else { if (metadataFilterColumns.contains(variableName)) { metadataSqlQuery = format("\"%s\" = %s", variableName, literalString); } - return new ClpExpression(format("%s: %s", variableName, literalString), metadataSqlQuery); + return new ClpExpression(format("%s: %s", variableName, literalString), metadataSqlQuery, ImmutableSet.of(variableName)); } } else if (operator.equals(NOT_EQUAL)) { if (literalType instanceof VarcharType) { - return new ClpExpression(format("NOT %s: \"%s\"", variableName, escapeKqlSpecialCharsForStringValue(literalString))); + return new ClpExpression( + format("NOT %s: \"%s\"", variableName, escapeKqlSpecialCharsForStringValue(literalString)), + ImmutableSet.of(variableName)); } else { if (metadataFilterColumns.contains(variableName)) { metadataSqlQuery = format("NOT \"%s\" = %s", variableName, literalString); } - return new ClpExpression(format("NOT %s: %s", variableName, literalString), metadataSqlQuery); + return new ClpExpression(format("NOT %s: %s", variableName, literalString), metadataSqlQuery, ImmutableSet.of(variableName)); } } else if (LOGICAL_BINARY_OPS_FILTER.contains(operator) && !(literalType instanceof VarcharType)) { if (metadataFilterColumns.contains(variableName)) { - metadataSqlQuery = format("\"%s\" %s %s", variableName, operator.getOperator(), literalString); + metadataSqlQuery = format("\"%s\" %s %s", variableName, operator.getOperator(), literalString, ImmutableSet.of(variableName)); } - return new ClpExpression(format("%s %s %s", variableName, operator.getOperator(), literalString), metadataSqlQuery); + return new ClpExpression( + format("%s %s %s", variableName, operator.getOperator(), literalString), metadataSqlQuery, ImmutableSet.of(variableName)); } return new ClpExpression(originalNode); } @@ -576,7 +587,7 @@ private Optional interpretSubstringEquality(SubstrInfo info, Stri result.append("?"); } result.append(targetString).append("*\""); - return Optional.of(new ClpExpression(result.toString())); + return Optional.of(new ClpExpression(result.toString(), ImmutableSet.of(info.variableName))); } } } @@ -590,11 +601,11 @@ private Optional interpretSubstringEquality(SubstrInfo info, Stri result.append("?"); } result.append(targetString).append("\""); - return Optional.of(new ClpExpression(result.toString())); + return Optional.of(new ClpExpression(result.toString(), ImmutableSet.of(info.variableName))); } if (start == -targetString.length()) { result.append(format("%s: \"*%s\"", info.variableName, targetString)); - return Optional.of(new ClpExpression(result.toString())); + return Optional.of(new ClpExpression(result.toString(), ImmutableSet.of(info.variableName))); } } } @@ -678,10 +689,12 @@ private ClpExpression handleAnd(SpecialFormExpression node) List remainingExpressions = new ArrayList<>(); boolean hasMetadataSql = false; boolean hasPushDownExpression = false; + ImmutableSet.Builder pushDownVariables = new ImmutableSet.Builder<>(); for (RowExpression argument : node.getArguments()) { ClpExpression expression = argument.accept(this, null); if (expression.getPushDownExpression().isPresent()) { hasPushDownExpression = true; + pushDownVariables.addAll(expression.getPushDownVariables()); queryBuilder.append(expression.getPushDownExpression().get()); queryBuilder.append(" AND "); if (expression.getMetadataSqlQuery().isPresent()) { @@ -702,18 +715,21 @@ else if (!remainingExpressions.isEmpty()) { return new ClpExpression( queryBuilder.substring(0, queryBuilder.length() - 5) + ")", hasMetadataSql ? metadataQueryBuilder.substring(0, metadataQueryBuilder.length() - 5) + ")" : null, - remainingExpressions.get(0)); + remainingExpressions.get(0), + pushDownVariables.build()); } else { return new ClpExpression( queryBuilder.substring(0, queryBuilder.length() - 5) + ")", hasMetadataSql ? metadataQueryBuilder.substring(0, metadataQueryBuilder.length() - 5) + ")" : null, - new SpecialFormExpression(node.getSourceLocation(), AND, BOOLEAN, remainingExpressions)); + new SpecialFormExpression(node.getSourceLocation(), AND, BOOLEAN, remainingExpressions), + pushDownVariables.build()); } } // Remove the last " AND " from the query return new ClpExpression(queryBuilder.substring(0, queryBuilder.length() - 5) + ")", - hasMetadataSql ? metadataQueryBuilder.substring(0, metadataQueryBuilder.length() - 5) + ")" : null); + hasMetadataSql ? metadataQueryBuilder.substring(0, metadataQueryBuilder.length() - 5) + ")" : null, + pushDownVariables.build()); } /** @@ -736,6 +752,7 @@ private ClpExpression handleOr(SpecialFormExpression node) queryBuilder.append("("); boolean allPushedDown = true; boolean hasAllMetadataSql = true; + ImmutableSet.Builder pushDownVariables = new ImmutableSet.Builder<>(); for (RowExpression argument : node.getArguments()) { ClpExpression expression = argument.accept(this, null); // Note: It is possible in the future that an expression cannot be pushed down as a KQL query, but can be @@ -746,6 +763,7 @@ private ClpExpression handleOr(SpecialFormExpression node) } queryBuilder.append(expression.getPushDownExpression().get()); queryBuilder.append(" OR "); + pushDownVariables.addAll(expression.getPushDownVariables()); if (hasAllMetadataSql && expression.getMetadataSqlQuery().isPresent()) { metadataQueryBuilder.append(expression.getMetadataSqlQuery().get()); metadataQueryBuilder.append(" OR "); @@ -758,7 +776,8 @@ private ClpExpression handleOr(SpecialFormExpression node) // Remove the last " OR " from the query return new ClpExpression( queryBuilder.substring(0, queryBuilder.length() - 4) + ")", - hasAllMetadataSql ? metadataQueryBuilder.substring(0, metadataQueryBuilder.length() - 4) + ")" : null); + hasAllMetadataSql ? metadataQueryBuilder.substring(0, metadataQueryBuilder.length() - 4) + ")" : null, + pushDownVariables.build()); } return new ClpExpression(node); } @@ -798,7 +817,7 @@ private ClpExpression handleIn(SpecialFormExpression node) } // Remove the last " OR " from the query - return new ClpExpression(queryBuilder.substring(0, queryBuilder.length() - 4) + ")"); + return new ClpExpression(queryBuilder.substring(0, queryBuilder.length() - 4) + ")", variable.getPushDownVariables()); } /** @@ -823,7 +842,7 @@ private ClpExpression handleIsNull(SpecialFormExpression node) } String variableName = expression.getPushDownExpression().get(); - return new ClpExpression(format("NOT %s: *", variableName)); + return new ClpExpression(format("NOT %s: *", variableName), expression.getPushDownVariables()); } /** @@ -885,7 +904,7 @@ private ClpExpression handleDereference(RowExpression expression) if (!baseString.getPushDownExpression().isPresent()) { return new ClpExpression(expression); } - return new ClpExpression(baseString.getPushDownExpression().get() + "." + fieldName); + return new ClpExpression(baseString.getPushDownExpression().get() + "." + fieldName, baseString.getPushDownVariables()); } /** @@ -925,6 +944,26 @@ public static boolean isClpCompatibleNumericType(Type type) || type instanceof DecimalType; } + private static String tryEnsureNanosecondTimestamp(Type type, String literalString) + { + if (type == TIMESTAMP) { + return ensureNanosecondTimestamp(TIMESTAMP, literalString); + } + else if (type == TIMESTAMP_MICROSECONDS) { + return ensureNanosecondTimestamp(TIMESTAMP_MICROSECONDS, literalString); + } + return literalString; + } + + private static String ensureNanosecondTimestamp(TimestampType type, String literalString) + { + long literalNumber = Long.parseLong(literalString); + long seconds = type.getEpochSecond(literalNumber); + long nanosecondFraction = type.getNanos(literalNumber); + long nanoseconds = SECONDS.toNanos(seconds) + nanosecondFraction; + return Long.toString(nanoseconds); + } + private static class SubstrInfo { String variableName; diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpPlanOptimizerProvider.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpPlanOptimizerProvider.java index b536c95ad216a..bdf50eb0fb709 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpPlanOptimizerProvider.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpPlanOptimizerProvider.java @@ -32,7 +32,10 @@ public class ClpPlanOptimizerProvider private final ClpSplitFilterProvider splitFilterProvider; @Inject - public ClpPlanOptimizerProvider(FunctionMetadataManager functionManager, StandardFunctionResolution functionResolution, ClpSplitFilterProvider splitFilterProvider) + public ClpPlanOptimizerProvider( + FunctionMetadataManager functionManager, + StandardFunctionResolution functionResolution, + ClpSplitFilterProvider splitFilterProvider) { this.functionManager = functionManager; this.functionResolution = functionResolution; diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpTopNSpec.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpTopNSpec.java new file mode 100644 index 0000000000000..de2f3ee2eab6c --- /dev/null +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/optimization/ClpTopNSpec.java @@ -0,0 +1,148 @@ +/* + * Licensed 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 com.facebook.presto.plugin.clp.optimization; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; + +import java.util.List; +import java.util.Objects; + +import static java.util.Objects.requireNonNull; + +/** + * Represents a Top-N specification for a query, including the limit of rows to return + * and the ordering of columns. + */ +public class ClpTopNSpec +{ + /** + * Enum representing the order direction: ascending or descending. + */ + public enum Order + { + ASC, + DESC + } + + /** + * Represents the ordering of one or more columns with a specified order (ASC or DESC). + */ + public static final class Ordering + { + private final List columns; + private final Order order; + + @JsonCreator + public Ordering( + @JsonProperty("columns") List columns, + @JsonProperty("order") Order order) + { + this.columns = requireNonNull(columns, "column is null"); + this.order = requireNonNull(order, "order is null"); + } + + @JsonProperty("columns") + public List getColumns() + { + return columns; + } + + @JsonProperty("order") + public Order getOrder() + { + return order; + } + + @Override + public int hashCode() + { + return Objects.hash(columns, order); + } + + @Override + public boolean equals(Object obj) + { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + Ordering other = (Ordering) obj; + return this.order == other.order && this.columns.equals(other.columns); + } + + @Override + public String toString() + { + return columns + ":" + order; + } + } + + private final long limit; + private final List orderings; + + @JsonCreator + public ClpTopNSpec( + @JsonProperty("limit") long limit, + @JsonProperty("orderings") List orderings) + { + if (limit <= 0) { + throw new IllegalArgumentException("limit must be > 0"); + } + if (orderings == null || orderings.isEmpty()) { + throw new IllegalArgumentException("orderings must be non-empty"); + } + this.limit = limit; + this.orderings = orderings; + } + + @JsonProperty("limit") + public long getLimit() + { + return limit; + } + + @JsonProperty("orderings") + public List getOrderings() + { + return orderings; + } + + @Override + public int hashCode() + { + return Objects.hash(limit, orderings); + } + + @Override + public boolean equals(Object obj) + { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + ClpTopNSpec other = (ClpTopNSpec) obj; + return this.limit == other.limit && this.orderings.equals(other.orderings); + } + + @Override + public String toString() + { + return "ClpTopNSpec (limit=" + limit + ", order=" + orderings + ")"; + } +} diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpMySqlSplitProvider.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpMySqlSplitProvider.java index 6b54218509c7f..13435e28a8b2c 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpMySqlSplitProvider.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpMySqlSplitProvider.java @@ -18,6 +18,7 @@ import com.facebook.presto.plugin.clp.ClpSplit; import com.facebook.presto.plugin.clp.ClpTableHandle; import com.facebook.presto.plugin.clp.ClpTableLayoutHandle; +import com.facebook.presto.plugin.clp.optimization.ClpTopNSpec; import com.google.common.collect.ImmutableList; import javax.inject.Inject; @@ -27,22 +28,26 @@ import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; +import java.util.ArrayList; import java.util.List; +import java.util.Optional; import static com.facebook.presto.plugin.clp.ClpSplit.SplitType.ARCHIVE; import static java.lang.String.format; +import static java.util.Comparator.comparingLong; public class ClpMySqlSplitProvider implements ClpSplitProvider { // Column names public static final String ARCHIVES_TABLE_COLUMN_ID = "id"; + public static final String ARCHIVES_TABLE_NUM_MESSAGES = "num_messages"; // Table suffixes public static final String ARCHIVES_TABLE_SUFFIX = "_archives"; // SQL templates - private static final String SQL_SELECT_ARCHIVES_TEMPLATE = format("SELECT `%s` FROM `%%s%%s%s` WHERE 1 = 1", ARCHIVES_TABLE_COLUMN_ID, ARCHIVES_TABLE_SUFFIX); + private static final String SQL_SELECT_ARCHIVES_TEMPLATE = format("SELECT * FROM `%%s%%s%s` WHERE 1 = 1", ARCHIVES_TABLE_SUFFIX); private static final Logger log = Logger.get(ClpMySqlSplitProvider.class); @@ -66,6 +71,7 @@ public List listSplits(ClpTableLayoutHandle clpTableLayoutHandle) { ImmutableList.Builder splits = new ImmutableList.Builder<>(); ClpTableHandle clpTableHandle = clpTableLayoutHandle.getTable(); + Optional topNSpecOptional = clpTableLayoutHandle.getTopN(); String tablePath = clpTableHandle.getTablePath(); String tableName = clpTableHandle.getSchemaTableName().getTableName(); String archivePathQuery = format(SQL_SELECT_ARCHIVES_TEMPLATE, config.getMetadataTablePrefix(), tableName); @@ -74,6 +80,25 @@ public List listSplits(ClpTableLayoutHandle clpTableLayoutHandle) String metadataFilterQuery = clpTableLayoutHandle.getMetadataSql().get(); archivePathQuery += " AND (" + metadataFilterQuery + ")"; } + + if (topNSpecOptional.isPresent()) { + ClpTopNSpec topNSpec = topNSpecOptional.get(); + // Only handles one range metadata column for now + ClpTopNSpec.Ordering ordering = topNSpec.getOrderings().get(0); + String col = ordering.getColumns().get(ordering.getColumns().size() - 1); + String dir = (ordering.getOrder() == ClpTopNSpec.Order.ASC) ? "ASC" : "DESC"; + archivePathQuery += " ORDER BY " + "`" + col + "` " + dir; + + List archiveMetaList = fetchArchiveMeta(archivePathQuery, ordering); + List selected = selectTopNArchives(archiveMetaList, topNSpec.getLimit(), ordering.getOrder()); + + for (ArchiveMeta a : selected) { + splits.add(new ClpSplit(tablePath + "/" + a.id, ARCHIVE, clpTableLayoutHandle.getKqlQuery())); + } + ImmutableList result = splits.build(); + log.debug("Number of splits: %s", result.size()); + return result; + } log.debug("Query for archive: %s", archivePathQuery); try (Connection connection = getConnection()) { @@ -105,4 +130,171 @@ private Connection getConnection() } return connection; } + + /** + * Fetches archive metadata from the database. + * + * @param query SQL query string that selects the archives + * @param ordering The top-N ordering specifying which columns contain lowerBound/upperBound + * @return List of ArchiveMeta objects representing archive metadata + */ + private List fetchArchiveMeta(String query, ClpTopNSpec.Ordering ordering) + { + List list = new ArrayList<>(); + try (Connection connection = getConnection(); + PreparedStatement stmt = connection.prepareStatement(query); + ResultSet rs = stmt.executeQuery()) { + while (rs.next()) { + list.add(new ArchiveMeta( + rs.getString(ARCHIVES_TABLE_COLUMN_ID), + rs.getLong(ordering.getColumns().get(0)), + rs.getLong(ordering.getColumns().get(1)), + rs.getLong(ARCHIVES_TABLE_NUM_MESSAGES))); + } + } + catch (SQLException e) { + log.warn("Database error while fetching archive metadata: %s", e); + } + return list; + } + + /** + * Selects the set of archives that must be scanned to guarantee the top-N results by timestamp + * (ASC or DESC), given only archive ranges and message counts. + *
    + *
  • Merges overlapping archives into groups (union of time ranges).
    • + *
  • For DESC: always include the newest group, then add older ones until their total + * message counts cover the limit.
    • + *
  • For ASC: symmetric — start from the oldest, then add newer ones.
    • + *
+ + * @param archives list of archives with [lowerBound, upperBound, messageCount] + * @param limit number of messages requested + * @param order ASC (earliest first) or DESC (latest first) + * @return archives that must be scanned + */ + private static List selectTopNArchives(List archives, long limit, ClpTopNSpec.Order order) + { + if (archives == null || archives.isEmpty() || limit <= 0) { + return ImmutableList.of(); + } + + // 1) Merge overlaps into groups + List groups = toArchiveGroups(archives); + + // 2) Pick minimal set of groups per order, then return all member archives + List selected = new ArrayList<>(); + if (order == ClpTopNSpec.Order.DESC) { + // newest group index + int k = groups.size() - 1; + + // must include newest group + selected.addAll(groups.get(k).members); + + // assume worst case: newest contributes 0 after filter; cover limit from older groups + long coveredByOlder = 0; + for (int i = k - 1; i >= 0 && coveredByOlder < limit; --i) { + selected.addAll(groups.get(i).members); + coveredByOlder += groups.get(i).count; + } + } + else { + // oldest group index + int k = 0; + + // must include oldest group + selected.addAll(groups.get(k).members); + + // assume worst case: oldest contributes 0; cover limit from newer groups + long coveredByNewer = 0; + for (int i = k + 1; i < groups.size() && coveredByNewer < limit; ++i) { + selected.addAll(groups.get(i).members); + coveredByNewer += groups.get(i).count; + } + } + + return selected; + } + + /** + * Groups overlapping archives into non-overlapping archive groups. + * + * @param archives archives sorted by lowerBound + * @return merged groups + */ + private static List toArchiveGroups(List archives) + { + List sorted = new ArrayList<>(archives); + sorted.sort(comparingLong((ArchiveMeta a) -> a.lowerBound) + .thenComparingLong(a -> a.upperBound)); + + List groups = new ArrayList<>(); + ArchiveGroup cur = null; + + for (ArchiveMeta a : sorted) { + if (cur == null) { + cur = startArchiveGroup(a); + } + else if (overlaps(cur, a)) { + // extend current group + cur.end = Math.max(cur.end, a.upperBound); + cur.count += a.messageCount; + cur.members.add(a); + } + else { + // finalize current, start a new one + groups.add(cur); + cur = startArchiveGroup(a); + } + } + if (cur != null) { + groups.add(cur); + } + return groups; + } + + private static ArchiveGroup startArchiveGroup(ArchiveMeta a) + { + ArchiveGroup group = new ArchiveGroup(); + group.begin = a.lowerBound; + group.end = a.upperBound; + group.count = a.messageCount; + group.members.add(a); + return group; + } + + private static boolean overlaps(ArchiveGroup cur, ArchiveMeta a) + { + return a.lowerBound <= cur.end && a.upperBound >= cur.begin; + } + + /** + * Represents metadata of an archive, including its ID, timestamp bounds, and message count. + */ + private static class ArchiveMeta + { + final String id; + final long lowerBound; + final long upperBound; + final long messageCount; + + ArchiveMeta(String id, long lowerBound, long upperBound, long messageCount) + { + this.id = id; + this.lowerBound = lowerBound; + this.upperBound = upperBound; + this.messageCount = messageCount; + } + } + + /** + * Represents a group of overlapping archives treated as one logical unit. + */ + private static final class ArchiveGroup + { + long begin; + long end; + long count; + final List members = new ArrayList<>(); + } } diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpPinotSplitProvider.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpPinotSplitProvider.java new file mode 100644 index 0000000000000..5b2df4a71dabc --- /dev/null +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/ClpPinotSplitProvider.java @@ -0,0 +1,356 @@ +/* + * Licensed 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 com.facebook.presto.plugin.clp.split; + +import com.facebook.airlift.log.Logger; +import com.facebook.presto.plugin.clp.ClpConfig; +import com.facebook.presto.plugin.clp.ClpSplit; +import com.facebook.presto.plugin.clp.ClpTableHandle; +import com.facebook.presto.plugin.clp.ClpTableLayoutHandle; +import com.facebook.presto.plugin.clp.optimization.ClpTopNSpec; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.collect.ImmutableList; + +import javax.inject.Inject; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.HttpURLConnection; +import java.net.MalformedURLException; +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.Optional; + +import static com.facebook.presto.plugin.clp.ClpSplit.SplitType; +import static com.facebook.presto.plugin.clp.ClpSplit.SplitType.ARCHIVE; +import static com.facebook.presto.plugin.clp.ClpSplit.SplitType.IR; +import static java.lang.String.format; +import static java.util.Comparator.comparingLong; +import static java.util.Objects.requireNonNull; +import static java.util.concurrent.TimeUnit.SECONDS; + +public class ClpPinotSplitProvider + implements ClpSplitProvider +{ + private static final Logger log = Logger.get(ClpPinotSplitProvider.class); + private static final String SQL_SELECT_SPLITS_TEMPLATE = "SELECT tpath FROM %s WHERE 1 = 1 AND (%s) LIMIT 999999"; + private static final String SQL_SELECT_SPLIT_META_TEMPLATE = "SELECT tpath, creationtime, lastmodifiedtime, num_messages FROM %s WHERE 1 = 1 AND (%s) ORDER BY %s %s LIMIT 999999"; + private final ClpConfig config; + private final URL pinotDatabaseUrl; + + @Inject + public ClpPinotSplitProvider(ClpConfig config) + { + this.config = requireNonNull(config, "config is null"); + try { + this.pinotDatabaseUrl = new URL(config.getMetadataDbUrl() + "/query/sql"); + } + catch (MalformedURLException e) { + throw new IllegalArgumentException( + format("Invalid Pinot database URL: %s/query/sql", config.getMetadataDbUrl()), e); + } + } + + @Override + public List listSplits(ClpTableLayoutHandle clpTableLayoutHandle) + { + ClpTableHandle clpTableHandle = clpTableLayoutHandle.getTable(); + Optional topNSpecOptional = clpTableLayoutHandle.getTopN(); + String tableName = clpTableHandle.getSchemaTableName().getTableName(); + try { + ImmutableList.Builder splits = new ImmutableList.Builder<>(); + if (topNSpecOptional.isPresent()) { + ClpTopNSpec topNSpec = topNSpecOptional.get(); + // Only handles one range metadata column for now (first ordering) + ClpTopNSpec.Ordering ordering = topNSpec.getOrderings().get(0); + // Get the last column in the ordering (the primary sort column for nested fields) + String col = ordering.getColumns().get(ordering.getColumns().size() - 1); + String dir = (ordering.getOrder() == ClpTopNSpec.Order.ASC) ? "ASC" : "DESC"; + String splitMetaQuery = format(SQL_SELECT_SPLIT_META_TEMPLATE, tableName, clpTableLayoutHandle.getMetadataSql().orElse("1 = 1"), col, dir); + List archiveMetaList = fetchArchiveMeta(splitMetaQuery, ordering); + List selected = selectTopNArchives(archiveMetaList, topNSpec.getLimit(), ordering.getOrder()); + + for (ArchiveMeta a : selected) { + String splitPath = a.id; + splits.add(new ClpSplit(splitPath, determineSplitType(splitPath), clpTableLayoutHandle.getKqlQuery())); + } + + List filteredSplits = splits.build(); + log.debug("Number of topN filtered splits: %s", filteredSplits.size()); + return filteredSplits; + } + + String splitQuery = format(SQL_SELECT_SPLITS_TEMPLATE, tableName, clpTableLayoutHandle.getMetadataSql().orElse("1 = 1")); + List splitRows = getQueryResult(pinotDatabaseUrl, splitQuery); + for (JsonNode row : splitRows) { + String splitPath = row.elements().next().asText(); + splits.add(new ClpSplit(splitPath, determineSplitType(splitPath), clpTableLayoutHandle.getKqlQuery())); + } + + List filteredSplits = splits.build(); + log.debug("Number of filtered splits: %s", filteredSplits.size()); + return filteredSplits; + } + catch (Exception e) { + log.error(e, "Failed to list splits for table %s", tableName); + return Collections.emptyList(); + } + } + + /** + * Fetches archive metadata from the database. + * + * @param query SQL query string that selects the archives + * @param ordering The top-N ordering specifying which columns contain lowerBound/upperBound + * @return List of ArchiveMeta objects representing archive metadata + */ + private List fetchArchiveMeta(String query, ClpTopNSpec.Ordering ordering) + { + ImmutableList.Builder archiveMetas = new ImmutableList.Builder<>(); + List rows = getQueryResult(pinotDatabaseUrl, query); + for (JsonNode row : rows) { + archiveMetas.add(new ArchiveMeta( + row.get(0).asText(), + row.get(1).asLong(), + row.get(2).asLong(), + row.get(3).asLong())); + } + return archiveMetas.build(); + } + + /** + * Selects the set of archives that must be scanned to guarantee the top-N results by timestamp + * (ASC or DESC), given only archive ranges and message counts. + *
    + *
  • Merges overlapping archives into components (union of time ranges).
    • + *
  • For DESC: always include the newest component, then add older ones until their total + * message counts cover the limit.
    • + *
  • For ASC: symmetric — start from the oldest, then add newer ones.
    • + *
+ + * @param archives list of archives with [lowerBound, upperBound, messageCount] + * @param limit number of messages requested + * @param order ASC (earliest first) or DESC (latest first) + * @return archives that must be scanned + */ + private static List selectTopNArchives(List archives, long limit, ClpTopNSpec.Order order) + { + if (archives == null || archives.isEmpty() || limit <= 0) { + return ImmutableList.of(); + } + requireNonNull(order, "order is null"); + + // 1) Merge overlaps into groups + List groups = toArchiveGroups(archives); + + if (groups.isEmpty()) { + return ImmutableList.of(); + } + + // 2) Pick minimal set of groups per order, then return all member archives + List selected = new ArrayList<>(); + if (order == ClpTopNSpec.Order.DESC) { + // newest group index + int k = groups.size() - 1; + + // must include newest group + selected.addAll(groups.get(k).members); + + // assume worst case: newest contributes 0 after filter; cover limit from older groups + long coveredByOlder = 0; + for (int i = k - 1; i >= 0 && coveredByOlder < limit; --i) { + selected.addAll(groups.get(i).members); + coveredByOlder += groups.get(i).count; + } + } + else { + // oldest group index + int k = 0; + + // must include oldest group + selected.addAll(groups.get(k).members); + + // assume worst case: oldest contributes 0; cover limit from newer groups + long coveredByNewer = 0; + for (int i = k + 1; i < groups.size() && coveredByNewer < limit; ++i) { + selected.addAll(groups.get(i).members); + coveredByNewer += groups.get(i).count; + } + } + + return selected; + } + + /** + * Groups overlapping archives into non-overlapping archive groups. + * + * @param archives archives sorted by lowerBound + * @return merged components + */ + private static List toArchiveGroups(List archives) + { + List sorted = new ArrayList<>(archives); + sorted.sort(comparingLong((ArchiveMeta a) -> a.lowerBound) + .thenComparingLong(a -> a.upperBound)); + + List groups = new ArrayList<>(); + ArchiveGroup cur = null; + + for (ArchiveMeta a : sorted) { + if (cur == null) { + cur = startArchiveGroup(a); + } + else if (overlaps(cur, a)) { + // extend current component + cur.end = Math.max(cur.end, a.upperBound); + cur.count += a.messageCount; + cur.members.add(a); + } + else { + // finalize current, start a new one + groups.add(cur); + cur = startArchiveGroup(a); + } + } + if (cur != null) { + groups.add(cur); + } + return groups; + } + + private static ArchiveGroup startArchiveGroup(ArchiveMeta a) + { + ArchiveGroup group = new ArchiveGroup(); + group.begin = a.lowerBound; + group.end = a.upperBound; + group.count = a.messageCount; + group.members.add(a); + return group; + } + + private static boolean overlaps(ArchiveGroup cur, ArchiveMeta a) + { + return a.lowerBound <= cur.end && a.upperBound >= cur.begin; + } + + /** + * Determines the split type based on file path extension. + * + * @param splitPath the file path + * @return IR for .clp.zst files, ARCHIVE otherwise + */ + private static SplitType determineSplitType(String splitPath) + { + return splitPath.endsWith(".clp.zst") ? IR : ARCHIVE; + } + + private static List getQueryResult(URL url, String sql) + { + try { + HttpURLConnection conn = (HttpURLConnection) url.openConnection(); + conn.setRequestMethod("POST"); + conn.setRequestProperty("Content-Type", "application/json"); + conn.setRequestProperty("Accept", "application/json"); + conn.setDoOutput(true); + conn.setConnectTimeout((int) SECONDS.toMillis(5)); + conn.setReadTimeout((int) SECONDS.toMillis(30)); + + log.info("Executing Pinot query: %s", sql); + ObjectMapper mapper = new ObjectMapper(); + String body = format("{\"sql\": %s }", mapper.writeValueAsString(sql)); + try (OutputStream os = conn.getOutputStream()) { + os.write(body.getBytes(StandardCharsets.UTF_8)); + } + + int code = conn.getResponseCode(); + InputStream is = (code >= 200 && code < 300) ? conn.getInputStream() : conn.getErrorStream(); + if (is == null) { + throw new IOException("Pinot HTTP " + code + " with empty body"); + } + + JsonNode root; + try (InputStream in = is) { + root = mapper.readTree(in); + } + JsonNode resultTable = root.get("resultTable"); + if (resultTable == null) { + throw new IllegalStateException("Pinot query response missing 'resultTable' field"); + } + JsonNode rows = resultTable.get("rows"); + if (rows == null) { + throw new IllegalStateException("Pinot query response missing 'rows' field in resultTable"); + } + ImmutableList.Builder resultBuilder = ImmutableList.builder(); + for (Iterator it = rows.elements(); it.hasNext(); ) { + JsonNode row = it.next(); + resultBuilder.add(row); + } + List results = resultBuilder.build(); + log.debug("Number of results: %s", results.size()); + return results; + } + catch (IOException e) { + log.error(e, "IO error executing Pinot query: %s", sql); + return Collections.emptyList(); + } + catch (Exception e) { + log.error(e, "Unexpected error executing Pinot query: %s", sql); + return Collections.emptyList(); + } + } + + /** + * Represents metadata of an archive, including its ID, timestamp bounds, and message count. + */ + private static final class ArchiveMeta + { + private final String id; + private final long lowerBound; + private final long upperBound; + private final long messageCount; + + ArchiveMeta(String id, long lowerBound, long upperBound, long messageCount) + { + this.id = requireNonNull(id, "id is null"); + if (lowerBound > upperBound) { + throw new IllegalArgumentException( + format("Invalid archive bounds: lowerBound (%d) > upperBound (%d)", lowerBound, upperBound)); + } + if (messageCount < 0) { + throw new IllegalArgumentException( + format("Invalid message count: %d (must be >= 0)", messageCount)); + } + this.lowerBound = lowerBound; + this.upperBound = upperBound; + this.messageCount = messageCount; + } + } + + /** + * Represents a group of overlapping archives treated as one logical unit. + */ + private static final class ArchiveGroup + { + long begin; + long end; + long count; + final List members = new ArrayList<>(); + } +} diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/filter/ClpMySqlSplitFilterProvider.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/filter/ClpMySqlSplitFilterProvider.java index 31d24fd4df71c..4bec8a79c9eed 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/filter/ClpMySqlSplitFilterProvider.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/filter/ClpMySqlSplitFilterProvider.java @@ -15,6 +15,7 @@ import com.facebook.presto.plugin.clp.ClpConfig; import com.fasterxml.jackson.annotation.JsonProperty; +import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.inject.Inject; @@ -85,6 +86,30 @@ public String remapSplitFilterPushDownExpression(String scope, String pushDownEx return remappedSql; } + @Override + public List remapColumnName(String scope, String columnName) + { + String[] splitScope = scope.split("\\."); + + Map mappings = new HashMap<>(getAllMappingsFromFilters(filterMap.get(splitScope[0]))); + + if (1 < splitScope.length) { + mappings.putAll(getAllMappingsFromFilters(filterMap.get(splitScope[0] + "." + splitScope[1]))); + } + + if (3 == splitScope.length) { + mappings.putAll(getAllMappingsFromFilters(filterMap.get(scope))); + } + + if (mappings.containsKey(columnName)) { + ClpMySqlCustomSplitFilterOptions.RangeMapping value = mappings.get(columnName); + return ImmutableList.of(value.lowerBound, value.upperBound); + } + else { + return ImmutableList.of(columnName); + } + } + @Override protected Class getCustomSplitFilterOptionsClass() { diff --git a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/filter/ClpSplitFilterProvider.java b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/filter/ClpSplitFilterProvider.java index 0609843aaf22f..7f19a5296b801 100644 --- a/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/filter/ClpSplitFilterProvider.java +++ b/presto-clp/src/main/java/com/facebook/presto/plugin/clp/split/filter/ClpSplitFilterProvider.java @@ -91,20 +91,36 @@ public ClpSplitFilterProvider(ClpConfig config) */ public abstract String remapSplitFilterPushDownExpression(String scope, String pushDownExpression); + /** + * Rewrites {@code columnName} to remap column names based on the {@code "customOptions"} for + * the given scope. + *

+ * {@code scope} follows the format {@code catalog[.schema][.table]}, and determines which + * column mappings to apply, since mappings from more specific scopes (e.g., table-level) + * override or supplement those from broader scopes (e.g., catalog-level). For each scope + * (catalog, schema, table), this method collects all mappings defined in + * {@code "customOptions"}. + * + * @param scope the scope of the column mapping + * @param columnName the column name to be remapped + * @return the remapped column names + */ + public abstract List remapColumnName(String scope, String columnName); + /** * Checks for the given table, if {@code splitFilterPushDownExpression} contains all required * fields. * * @param tableScopeSet the set of scopes of the tables that are being queried - * @param splitFilterPushDownExpression the expression to be checked + * @param pushDownVariables the set of variables being pushed down */ - public void checkContainsRequiredFilters(Set tableScopeSet, String splitFilterPushDownExpression) + public void checkContainsRequiredFilters(Set tableScopeSet, Set pushDownVariables) { boolean hasRequiredSplitFilterColumns = true; ImmutableList.Builder notFoundListBuilder = ImmutableList.builder(); for (String tableScope : tableScopeSet) { for (String columnName : getRequiredColumnNames(tableScope)) { - if (!splitFilterPushDownExpression.contains(columnName)) { + if (!pushDownVariables.contains(columnName)) { hasRequiredSplitFilterColumns = false; notFoundListBuilder.add(columnName); } diff --git a/presto-clp/src/test/java/com/facebook/presto/plugin/clp/ClpMetadataDbSetUp.java b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/ClpMetadataDbSetUp.java index d1d0ee6964c8e..ee207f9864004 100644 --- a/presto-clp/src/test/java/com/facebook/presto/plugin/clp/ClpMetadataDbSetUp.java +++ b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/ClpMetadataDbSetUp.java @@ -37,6 +37,7 @@ import static com.facebook.presto.plugin.clp.metadata.ClpMySqlMetadataProvider.DATASETS_TABLE_COLUMN_NAME; import static com.facebook.presto.plugin.clp.metadata.ClpMySqlMetadataProvider.DATASETS_TABLE_SUFFIX; import static com.facebook.presto.plugin.clp.split.ClpMySqlSplitProvider.ARCHIVES_TABLE_COLUMN_ID; +import static com.facebook.presto.plugin.clp.split.ClpMySqlSplitProvider.ARCHIVES_TABLE_NUM_MESSAGES; import static com.facebook.presto.plugin.clp.split.ClpMySqlSplitProvider.ARCHIVES_TABLE_SUFFIX; import static java.lang.String.format; import static java.util.UUID.randomUUID; @@ -139,26 +140,30 @@ public static ClpMySqlSplitProvider setupSplit(DbHandle dbHandle, Map("msg.timestamp", Integer), + new Pair<>("city.Name", ClpString), + new Pair<>("city.Region.Id", Integer), + new Pair<>("city.Region.Name", VarString), + new Pair<>("fare", Float), + new Pair<>("isHoliday", Boolean)))); + + splitProvider = setupSplit(dbHandle, + ImmutableMap.of( + tableName, + ImmutableList.of( + new ClpMetadataDbSetUp.ArchivesTableRow("0", 100, 0, 100), + new ClpMetadataDbSetUp.ArchivesTableRow("1", 100, 50, 150), + new ClpMetadataDbSetUp.ArchivesTableRow("2", 100, 100, 200), + new ClpMetadataDbSetUp.ArchivesTableRow("3", 100, 201, 300), + new ClpMetadataDbSetUp.ArchivesTableRow("4", 100, 301, 400)))); + + URL resource = getClass().getClassLoader().getResource("test-topn-split-filter.json"); + if (resource == null) { + log.error("test-topn-split-filter.json not found in resources"); + return; + } + + String filterConfigPath; + try { + filterConfigPath = Paths.get(resource.toURI()).toAbsolutePath().toString(); + } + catch (URISyntaxException e) { + log.error("test-topn-split-filter.json not found in resources"); + return; + } + + localQueryRunner = new LocalQueryRunner(defaultSession); + localQueryRunner.createCatalog("clp", new ClpConnectorFactory(), ImmutableMap.of( + "clp.metadata-db-url", format(METADATA_DB_URL_TEMPLATE, dbHandle.getDbPath()), + "clp.metadata-db-user", METADATA_DB_USER, + "clp.metadata-db-password", METADATA_DB_PASSWORD, + "clp.metadata-table-prefix", METADATA_DB_TABLE_PREFIX)); + localQueryRunner.getMetadata().registerBuiltInFunctions(extractFunctions(new ClpPlugin().getFunctions())); + functionAndTypeManager = localQueryRunner.getMetadata().getFunctionAndTypeManager(); + functionResolution = new FunctionResolution(functionAndTypeManager.getFunctionAndTypeResolver()); + splitFilterProvider = new ClpMySqlSplitFilterProvider(new ClpConfig().setSplitFilterConfig(filterConfigPath)); + planNodeIdAllocator = new PlanNodeIdAllocator(); + variableAllocator = new VariableAllocator(); + } + + @AfterMethod + public void tearDown() + { + localQueryRunner.close(); + ClpMetadataDbSetUp.tearDown(dbHandle); + } + + @Test + public void test() + { + testTopNQueryPlanAndSplits( + "SELECT * FROM test WHERE msg.timestamp > 120 AND msg.timestamp < 240 ORDER BY msg.timestamp DESC LIMIT 100", + "(msg.timestamp > 120 AND msg.timestamp < 240)", + "(end_timestamp > 120 AND begin_timestamp < 240)", + 100, + DESC, + ImmutableSet.of("1", "2", "3")); + + testTopNQueryPlanAndSplits( + "SELECT * FROM test WHERE msg.timestamp > 120 AND msg.timestamp < 240 ORDER BY msg.timestamp ASC LIMIT 50", + "(msg.timestamp > 120 AND msg.timestamp < 240)", + "(end_timestamp > 120 AND begin_timestamp < 240)", + 50, + ASC, + ImmutableSet.of("1", "2", "3")); + + testTopNQueryPlanAndSplits( + "SELECT * FROM test WHERE msg.timestamp >= 180 AND msg.timestamp <= 260 ORDER BY msg.timestamp DESC LIMIT 100", + "(msg.timestamp >= 180 AND msg.timestamp <= 260)", + "(end_timestamp >= 180 AND begin_timestamp <= 260)", + 100, + DESC, + ImmutableSet.of("2", "3")); + + testTopNQueryPlanAndSplits( + "SELECT * FROM test WHERE msg.timestamp > 250 AND msg.timestamp < 290 ORDER BY msg.timestamp DESC LIMIT 10", + "(msg.timestamp > 250 AND msg.timestamp < 290)", + "(end_timestamp > 250 AND begin_timestamp < 290)", + 10, + DESC, + ImmutableSet.of("3")); + + testTopNQueryPlanAndSplits( + "SELECT * FROM test WHERE msg.timestamp > 1000 AND msg.timestamp < 1100 ORDER BY msg.timestamp DESC LIMIT 10", + "(msg.timestamp > 1000 AND msg.timestamp < 1100)", + "(end_timestamp > 1000 AND begin_timestamp < 1100)", + 10, + DESC, + ImmutableSet.of()); + + testTopNQueryPlanAndSplits( + "SELECT * FROM test WHERE msg.timestamp <= 300 ORDER BY msg.timestamp DESC LIMIT 1000", + "msg.timestamp <= 300", + "begin_timestamp <= 300", + 1000, + DESC, + ImmutableSet.of("0", "1", "2", "3")); + + testTopNQueryPlanAndSplits( + "SELECT * FROM test WHERE msg.timestamp <= 400 ORDER BY msg.timestamp DESC LIMIT 100", + "msg.timestamp <= 400", + "begin_timestamp <= 400", + 100, + DESC, + ImmutableSet.of("3", "4")); + } + + private void testTopNQueryPlanAndSplits(String sql, String kql, String metadataSql, long limit, Order order, Set splitIds) + { + TransactionId transactionId = localQueryRunner.getTransactionManager().beginTransaction(false); + Session session = testSessionBuilder().setCatalog("clp").setSchema("default").setTransactionId(transactionId).build(); + + Plan plan = localQueryRunner.createPlan( + session, + sql, + WarningCollector.NOOP); + ClpComputePushDown optimizer = new ClpComputePushDown(functionAndTypeManager, functionResolution, splitFilterProvider); + PlanNode optimizedPlan = optimizer.optimize(plan.getRoot(), session.toConnectorSession(), variableAllocator, planNodeIdAllocator); + PlanNode optimizedPlanWithUniqueId = freshenIds(optimizedPlan, new PlanNodeIdAllocator()); + + ClpTableLayoutHandle clpTableLayoutHandle = new ClpTableLayoutHandle( + table, + Optional.of(kql), + Optional.of(metadataSql), + true, + Optional.of(new ClpTopNSpec( + limit, + ImmutableList.of(new ClpTopNSpec.Ordering(ImmutableList.of("begin_timestamp", "end_timestamp"), order))))); + + PlanAssert.assertPlan( + session, + localQueryRunner.getMetadata(), + (node, sourceStats, lookup, s, types) -> PlanNodeStatsEstimate.unknown(), + new Plan(optimizedPlanWithUniqueId, plan.getTypes(), StatsAndCosts.empty()), + anyTree( + ClpTableScanMatcher.clpTableScanPattern( + clpTableLayoutHandle, + ImmutableSet.of( + city, + fare, + isHoliday, + new ClpColumnHandle( + "msg", + RowType.from(ImmutableList.of(new RowType.Field(Optional.of("timestamp"), BIGINT)))))))); + + assertEquals( + ImmutableSet.copyOf(splitProvider.listSplits(clpTableLayoutHandle)), + splitIds.stream() + .map(id -> new ClpSplit("/tmp/archives/test/" + id, ARCHIVE, Optional.of(kql))) + .collect(ImmutableSet.toImmutableSet())); + } + + /** + * Recursively rebuilds a query plan tree so that every {@link PlanNode} has a fresh, unique ID. + *

+ * This utility is mainly for testing, to avoid ID collisions that can occur when + * localQueryRunner.createPlan() and a custom optimizer each use separate + * {@link PlanNodeIdAllocator}s that start at the same seed, producing duplicate IDs. + * + * @param root the root of the plan + * @param idAlloc the plan node ID allocator + * @return the plan with a fresh, unique IDs. + */ + private static PlanNode freshenIds(PlanNode root, PlanNodeIdAllocator idAlloc) + { + return SimplePlanRewriter.rewriteWith(new SimplePlanRewriter() { + @Override + public PlanNode visitOutput(OutputNode node, RewriteContext ctx) + { + PlanNode src = ctx.rewrite(node.getSource(), null); + return new OutputNode( + node.getSourceLocation(), + idAlloc.getNextId(), + src, + node.getColumnNames(), + node.getOutputVariables()); + } + + @Override + public PlanNode visitExchange(ExchangeNode node, RewriteContext ctx) + { + List newSources = node.getSources().stream() + .map(s -> ctx.rewrite(s, null)) + .collect(com.google.common.collect.ImmutableList.toImmutableList()); + + return new ExchangeNode( + node.getSourceLocation(), + idAlloc.getNextId(), + node.getType(), + node.getScope(), + node.getPartitioningScheme(), + newSources, + node.getInputs(), + node.isEnsureSourceOrdering(), + node.getOrderingScheme()); + } + + @Override + public PlanNode visitProject(ProjectNode node, RewriteContext ctx) + { + PlanNode src = ctx.rewrite(node.getSource(), null); + return new ProjectNode(idAlloc.getNextId(), src, node.getAssignments()); + } + + @Override + public PlanNode visitFilter(FilterNode node, RewriteContext ctx) + { + PlanNode src = ctx.rewrite(node.getSource(), null); + return new FilterNode(node.getSourceLocation(), idAlloc.getNextId(), src, node.getPredicate()); + } + + @Override + public PlanNode visitTopN(TopNNode node, RewriteContext ctx) + { + PlanNode src = ctx.rewrite(node.getSource(), null); + return new TopNNode( + node.getSourceLocation(), + idAlloc.getNextId(), + src, + node.getCount(), + node.getOrderingScheme(), + node.getStep()); + } + + @Override + public PlanNode visitTableScan(TableScanNode node, RewriteContext ctx) + { + return new TableScanNode( + node.getSourceLocation(), + idAlloc.getNextId(), + node.getTable(), + node.getOutputVariables(), + node.getAssignments()); + } + + @Override + public PlanNode visitPlan(PlanNode node, RewriteContext ctx) + { + List newChildren = node.getSources().stream() + .map(ch -> ctx.rewrite(ch, null)) + .collect(com.google.common.collect.ImmutableList.toImmutableList()); + return node.replaceChildren(newChildren); + } + }, root, null); + } + + private static final class ClpTableScanMatcher + implements Matcher + { + private final ClpTableLayoutHandle expectedLayoutHandle; + private final Set expectedColumns; + + private ClpTableScanMatcher(ClpTableLayoutHandle expectedLayoutHandle, Set expectedColumns) + { + this.expectedLayoutHandle = expectedLayoutHandle; + this.expectedColumns = expectedColumns; + } + + static PlanMatchPattern clpTableScanPattern(ClpTableLayoutHandle layoutHandle, Set columns) + { + return node(TableScanNode.class).with(new ClpTableScanMatcher(layoutHandle, columns)); + } + + @Override + public boolean shapeMatches(PlanNode node) + { + return node instanceof TableScanNode; + } + + @Override + public MatchResult detailMatches( + PlanNode node, + StatsProvider stats, + Session session, + Metadata metadata, + SymbolAliases symbolAliases) + { + checkState(shapeMatches(node), "Plan testing framework error: shapeMatches returned false"); + TableScanNode tableScanNode = (TableScanNode) node; + ClpTableLayoutHandle actualLayoutHandle = (ClpTableLayoutHandle) tableScanNode.getTable().getLayout().get(); + + // Check layout handle + if (!expectedLayoutHandle.equals(actualLayoutHandle)) { + return NO_MATCH; + } + + // Check assignments contain expected columns + Map actualAssignments = tableScanNode.getAssignments(); + Set actualColumns = new HashSet<>(actualAssignments.values()); + + if (!expectedColumns.equals(actualColumns)) { + return NO_MATCH; + } + + SymbolAliases.Builder aliasesBuilder = SymbolAliases.builder(); + for (VariableReferenceExpression variable : tableScanNode.getOutputVariables()) { + aliasesBuilder.put(variable.getName(), new SymbolReference(variable.getName())); + } + + return match(aliasesBuilder.build()); + } + } +} diff --git a/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpYamlMetadata.java b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpYamlMetadata.java new file mode 100644 index 0000000000000..e5a45d05692ae --- /dev/null +++ b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/TestClpYamlMetadata.java @@ -0,0 +1,129 @@ +/* + * Licensed 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 com.facebook.presto.plugin.clp; + +import com.facebook.presto.plugin.clp.metadata.ClpMetadataProvider; +import com.facebook.presto.plugin.clp.metadata.ClpYamlMetadataProvider; +import com.facebook.presto.plugin.clp.split.ClpPinotSplitProvider; +import com.facebook.presto.plugin.clp.split.ClpSplitProvider; +import com.facebook.presto.spi.ColumnMetadata; +import com.facebook.presto.spi.ConnectorTableMetadata; +import com.facebook.presto.spi.SchemaTableName; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import org.testng.annotations.BeforeTest; +import org.testng.annotations.Test; + +import java.util.HashSet; +import java.util.List; +import java.util.Optional; + +import static com.facebook.presto.plugin.clp.ClpConfig.MetadataProviderType.YAML; +import static com.facebook.presto.plugin.clp.ClpMetadata.DEFAULT_SCHEMA_NAME; +import static com.facebook.presto.testing.TestingConnectorSession.SESSION; +import static org.testng.Assert.assertEquals; + +public class TestClpYamlMetadata +{ + private static final String PINOT_BROKER_URL = "http://localhost:8099"; + private static final String METADATA_YAML_PATH = "/home/xiaochong-dev/presto-e2e/pinot/tables-schema.yaml"; + private static final String TABLE_NAME = "cockroachdb"; + private ClpMetadata metadata; + private ClpSplitProvider clpSplitProvider; + + @BeforeTest + public void setUp() + { + ClpConfig config = new ClpConfig() + .setPolymorphicTypeEnabled(true) + .setMetadataDbUrl(PINOT_BROKER_URL) + .setMetadataProviderType(YAML) + .setMetadataYamlPath(METADATA_YAML_PATH); + ClpMetadataProvider metadataProvider = new ClpYamlMetadataProvider(config); + metadata = new ClpMetadata(config, metadataProvider); + clpSplitProvider = new ClpPinotSplitProvider(config); + } + + @Test + public void testListSchemaNames() + { + assertEquals(metadata.listSchemaNames(SESSION), ImmutableList.of(DEFAULT_SCHEMA_NAME)); + } + + @Test + public void testListTables() + { + ImmutableSet.Builder builder = ImmutableSet.builder(); + builder.add(new SchemaTableName(DEFAULT_SCHEMA_NAME, TABLE_NAME)); + assertEquals(new HashSet<>(metadata.listTables(SESSION, Optional.empty())), builder.build()); + } + + @Test + public void testListSplits() + { + ClpTableLayoutHandle layoutHandle = new ClpTableLayoutHandle( + new ClpTableHandle(new SchemaTableName(DEFAULT_SCHEMA_NAME, TABLE_NAME), ""), + Optional.empty(), + Optional.empty()); + List result = clpSplitProvider.listSplits(layoutHandle); + System.out.println("Hello world"); + } + + @Test + public void testGetTableMetadata() + { + ClpTableHandle clpTableHandle = (ClpTableHandle) metadata.getTableHandle(SESSION, new SchemaTableName(DEFAULT_SCHEMA_NAME, TABLE_NAME)); + ConnectorTableMetadata tableMetadata = metadata.getTableMetadata(SESSION, clpTableHandle); +// ImmutableSet columnMetadata = ImmutableSet.builder() +// .add(ColumnMetadata.builder() +// .setName("a_bigint") +// .setType(BIGINT) +// .setNullable(true) +// .build()) +// .add(ColumnMetadata.builder() +// .setName("a_varchar") +// .setType(VARCHAR) +// .setNullable(true) +// .build()) +// .add(ColumnMetadata.builder() +// .setName("b_double") +// .setType(DOUBLE) +// .setNullable(true) +// .build()) +// .add(ColumnMetadata.builder() +// .setName("b_varchar") +// .setType(VARCHAR) +// .setNullable(true) +// .build()) +// .add(ColumnMetadata.builder() +// .setName("c") +// .setType(RowType.from(ImmutableList.of( +// RowType.field("d", BOOLEAN), +// RowType.field("e", VARCHAR)))) +// .setNullable(true) +// .build()) +// .add(ColumnMetadata.builder() +// .setName("f") +// .setType(RowType.from(ImmutableList.of( +// RowType.field("g", +// RowType.from(ImmutableList.of( +// RowType.field("h", new ArrayType(VARCHAR)))))))) +// .setNullable(true) +// .build()) +// .build(); +// assertEquals(columnMetadata, ImmutableSet.copyOf(tableMetadata.getColumns())); + ImmutableSet actual = ImmutableSet.copyOf(tableMetadata.getColumns()); + System.out.println("Hello world"); + } +} diff --git a/presto-clp/src/test/java/com/facebook/presto/plugin/clp/split/filter/TestClpSplitFilterConfigCommon.java b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/split/filter/TestClpSplitFilterConfigCommon.java index 7a4058f617d0c..c6fe1cee8bec2 100644 --- a/presto-clp/src/test/java/com/facebook/presto/plugin/clp/split/filter/TestClpSplitFilterConfigCommon.java +++ b/presto-clp/src/test/java/com/facebook/presto/plugin/clp/split/filter/TestClpSplitFilterConfigCommon.java @@ -57,12 +57,8 @@ public void checkRequiredFilters() config.setSplitFilterConfig(filterConfigPath); ClpMySqlSplitFilterProvider filterProvider = new ClpMySqlSplitFilterProvider(config); Set testTableScopeSet = ImmutableSet.of(format("%s.%s", CONNECTOR_NAME, new SchemaTableName("default", "table_1"))); - assertThrows(PrestoException.class, () -> filterProvider.checkContainsRequiredFilters( - testTableScopeSet, - "(\"level\" >= 1 AND \"level\" <= 3)")); - filterProvider.checkContainsRequiredFilters( - testTableScopeSet, - "(\"msg.timestamp\" > 1234 AND \"msg.timestamp\" < 5678)"); + assertThrows(PrestoException.class, () -> filterProvider.checkContainsRequiredFilters(testTableScopeSet, ImmutableSet.of("level"))); + filterProvider.checkContainsRequiredFilters(testTableScopeSet, ImmutableSet.of("msg.timestamp")); } @Test diff --git a/presto-clp/src/test/resources/test-topn-split-filter.json b/presto-clp/src/test/resources/test-topn-split-filter.json new file mode 100644 index 0000000000000..53450716cb7b4 --- /dev/null +++ b/presto-clp/src/test/resources/test-topn-split-filter.json @@ -0,0 +1,14 @@ +{ + "clp.default.test": [ + { + "columnName": "msg.timestamp", + "customOptions": { + "rangeMapping": { + "lowerBound": "begin_timestamp", + "upperBound": "end_timestamp" + } + }, + "required": true + } + ] +} diff --git a/presto-native-execution/pom.xml b/presto-native-execution/pom.xml index 200ffa6834afc..adbd01ab4b917 100644 --- a/presto-native-execution/pom.xml +++ b/presto-native-execution/pom.xml @@ -267,6 +267,16 @@ + + + + org.yaml + snakeyaml + 2.1 + + + +