This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 1ff8aba81be [improvement](fe) Optimize external file COUNT scan
scheduling (#68726)
1ff8aba81be is described below
commit 1ff8aba81bedc59b463002c6f122a08e90a5c81f
Author: Oliveira <[email protected]>
AuthorDate: Tue Oct 13 15:19:57 2026 +0800
[improvement](fe) Optimize external file COUNT scan scheduling (#68726)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #68964
Problem Summary:
Repeated queries against a single remotely accessible file split can
repeatedly select the same backend when external file caching and
consistent-hash scheduling are disabled. The round-robin scheduler
reconstructed its candidate list from a host map, discarding the
query-local randomized order selected during initialization. This change
preserves that order for round-robin assignment and keeps local
scheduling and split redistribution inside the active backend
candidates.
For unfiltered, ungrouped `COUNT(*)` and `COUNT(1)` over File, HDFS,
HTTP, Local, and S3 TVFs, the existing file COUNT pushdown is now used
under the existing session switches. The instance reduction is limited
to a file TVF row-count scan with exactly one assigned static physical
split. Dynamic `split_source` entries, multiple physical splits in one
range envelope, and existing external-table COUNT paths retain their
original parallelism. DISTINCT, grouped aggregates, filtered plans,
merge stages, and stream/ingest TVFs remain outside this path.
The single-split CSV COUNT benchmark improved median QPS from 169.600 to
543.733 (3.206x) under the conditions below. Dynamic batch scans retain
baseline fragment-instance parallelism.
Stream Load schema-version locking has been removed from this PR: a
differential unit test reproduces the race on the pre-optimization
baseline. It is handled separately in #68964, including HTTP transaction
cleanup on table-lock timeout. This PR retains the external scan
scheduling and file TVF COUNT changes.
### Release note
Improve round-robin scheduling for external file scans and reduce
execution overhead for unfiltered file TVF row-count queries with a
single assigned static physical split.
### Check List (For Author)
- Test
- [x] Regression test
- [x] Unit Test
- [x] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
Validation performed:
- After separating the independent Stream Load schema-lock fix, the
standard focused FE UT reactor passed 33 tests:
`FederationBackendPolicyTest` (12), `FileTvfCountPushDownTest` (5),
`UnassignedScanSingleRemoteTableJobTest` (7), and the upstream
`StreamLoadHandlerTest` (9). Maven Checkstyle passed. This is functional
validation of head `0afd4837804`; the benchmark numbers below are from
the earlier measured revision.
- Standard full FE packaging and Checkstyle passed. Standard focused FE
UT reactor passed: `FederationBackendPolicyTest` (12),
`FileTvfCountPushDownTest` (4), and
`UnassignedScanSingleRemoteTableJobTest` (7), 23 tests total, zero
failures/errors. Parallelism tests cover dynamic split sources, multiple
physical splits in a single range envelope, external-table exclusion,
and actual local-shuffle instance assignment.
- New `test_file_tvf_count_pushdown` passed both framework snapshot
generation and normal comparison (1 suite, zero failures/fatal scripts).
It exercises CSV, JSON, Parquet, and ORC with scanner V1/V2 and pushdown
enabled/disabled, including COUNT star/one/nullable/DISTINCT, filters,
grouping, empty inputs, multiple files, and CSV headers/blank records/no
final newline. Zero-byte JSON is included alongside nonempty files
because standalone schema inference is unavailable.
- `test_s3_tvf_compression` passed against the personal MinIO equivalent
using the same public compression fixtures (1 suite, zero failures/fatal
scripts). The standard Aliyun OSS profile remains unvalidated because
valid credentials are unavailable; community CI should rerun it with its
configured credentials.
- Full Release BE build passed. Both FE versions were tested with
identical newly built BE binaries; the new four-format and compressed
TVF suites were also repeated successfully on this runtime.
- Single static split CSV (`COUNT(*)`, 1,000,000 rows, 64 clients):
three baseline QPS values were **165.533 / 170.800 / 169.600**;
candidate values were **449.133 / 544.967 / 543.733**. Median QPS
increased from **169.600 to 543.733 (3.206x)**. This measures the
combined scheduling and COUNT changes for this workload; it does not
establish a scheduling-only multiplier.
- Real dynamic Iceberg batch (`COUNT(value)`, 64 data files, 1,000,000
rows, expected count 500,000, 16 clients): baseline QPS **29.533 /
30.367 / 28.233**, candidate **30.167 / 29.767 / 29.267**; medians
**29.533 / 29.767 (1.008x)**. Distributed plans assigned **192 scan
fragment instances, 64 per BE**, for both versions in all rounds,
confirming dynamic sources keep their baseline instance parallelism.
These are assigned fragment instances, not active scanner-thread counts.
This throughput comparison is limited to the tested fixture and its
observed variability.
- All twelve measurements had verified COUNT results, zero errors, and
3/3 post-measurement backend health. Final artifact inspection found ten
connector/dependency JARs with an older API-version manifest because
size/mtime-based synchronization skipped them. ZIP entry comparisons
confirmed identical classes/resources; the Iceberg provider JAR itself
and FE core matched the intended builds. API gating reads the provider's
defining JAR, so the differing SPI dependency manifest is outside this
check. The measurements are therefore a comparison of the tested
workload paths, not a claim of identical complete FE packaging.
Unrelated ADBC/ES/Hudi/Trino providers were rejected during startup. A
subsequent checksum deployment matched all 2,213 FE artifact hashes and
all three BE binaries to the frozen build manifest. All previously
rejected providers and Iceberg loaded successfully; static
COUNT=1,000,000, dynamic nullable COUNT=500,000, and 3/3 health passed.
This deployment smoke did not rerun the performance measurements.
- Manual benchmark procedure: alternate baseline/candidate FE builds in
three rounds (B/C, C/B, B/C), restart all three isolated BEs after each
FE switch, require 3/3 health and the expected COUNT result, warm up for
10 seconds, then measure verified successful queries over 30 seconds.
Use identical Release BE binaries and session settings
(`parallel_pipeline_task_num=64`, file cache/consistent hash/SQL
cache/query cache disabled, COUNT pushdown and scanner V2 enabled). Use
64 clients for the single-split CSV and 16 for the dynamic Iceberg
batch; keep profiling disabled during measurement. All measurements
require zero errors. The three-BE restart prevents stale cached FE RPC
connections after version swaps; startup failures are excluded before
measurement.
- Benchmarks use three BEs on one shared host and report only the tested
workload; no equivalent multiplier is claimed for all queries or
customer deployments.
- Behavior changed:
- [ ] No.
- [x] Yes. Round-robin assignment preserves initialized candidate order;
preferred load subsets are honored by local scheduling and
redistribution; eligible file TVF COUNT scans use pushdown and one
instance per worker when that worker has exactly one static physical
split; dynamic sources and existing external-table COUNT parallelism
remain unchanged.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../datasource/scan/FederationBackendPolicy.java | 25 +-
.../glue/translator/PhysicalPlanTranslator.java | 59 ++-
.../worker/job/AbstractUnassignedScanJob.java | 8 +-
.../job/UnassignedScanSingleRemoteTableJob.java | 28 ++
.../glue/translator/FileTvfCountPushDownTest.java | 155 ++++++++
.../UnassignedScanSingleRemoteTableJobTest.java | 162 ++++++++
.../doris/planner/FederationBackendPolicyTest.java | 121 ++++++
.../tvf/test_file_tvf_count_pushdown.out | 428 +++++++++++++++++++++
.../tvf/test_s3_tvf_compression.out | 6 +
.../external_table_p0/tvf/tvf_count_data_empty.csv | 0
.../tvf/tvf_count_data_empty.json | 0
.../external_table_p0/tvf/tvf_count_data_empty.orc | Bin 0 -> 116 bytes
.../tvf/tvf_count_data_empty.parquet | Bin 0 -> 481 bytes
.../external_table_p0/tvf/tvf_count_data_more.csv | 4 +
.../external_table_p0/tvf/tvf_count_data_more.json | 4 +
.../external_table_p0/tvf/tvf_count_data_more.orc | Bin 0 -> 306 bytes
.../tvf/tvf_count_data_more.parquet | Bin 0 -> 1052 bytes
.../external_table_p0/tvf/tvf_count_data_rows.csv | 4 +
.../external_table_p0/tvf/tvf_count_data_rows.json | 4 +
.../external_table_p0/tvf/tvf_count_data_rows.orc | Bin 0 -> 306 bytes
.../tvf/tvf_count_data_rows.parquet | Bin 0 -> 1052 bytes
.../external_table_p0/tvf/tvf_count_header.csv | 6 +
.../cloud_p0/test_read_cluster_var_property.groovy | 4 +-
.../tvf/test_file_tvf_count_pushdown.groovy | 111 ++++++
.../tvf/test_s3_tvf_compression.groovy | 29 ++
25 files changed, 1145 insertions(+), 13 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java
index bb23481fd4d..4b4572b2ce7 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java
@@ -191,11 +191,12 @@ public class FederationBackendPolicy {
throw new UserException("No available backends for compute
group: " + computeGroup.toString());
}
}
+ backendMap.clear();
+
backendMap.putAll(backends.stream().collect(Collectors.groupingBy(Backend::getHost)));
for (Backend backend : backends) {
assignedWeightPerBackend.put(backend, 0L);
}
-
backendMap.putAll(backends.stream().collect(Collectors.groupingBy(Backend::getHost)));
try {
consistentHash = consistentHashCache.get(new
HashCacheKey(backends));
} catch (ExecutionException e) {
@@ -209,10 +210,21 @@ public class FederationBackendPolicy {
return selectedBackend;
}
- /** Replace only the round-robin order; candidate membership and backend
metadata remain unchanged. */
+ /**
+ * Replace the round-robin order with a non-empty subset of the
initialized candidates.
+ * Load backend selection may intentionally retain only preferred
candidates.
+ */
public void replaceBackendOrder(List<Backend> orderedBackends) {
+ Preconditions.checkState(nodeSelectionStrategy ==
NodeSelectionStrategy.ROUND_ROBIN,
+ "backend order can only be replaced for round-robin
selection");
+ Preconditions.checkArgument(!orderedBackends.isEmpty()
+ && new LinkedHashSet<>(orderedBackends).size() ==
orderedBackends.size()
+ && new
LinkedHashSet<>(backends).containsAll(orderedBackends),
+ "ordered backends must be a non-empty subset of the
initialized candidates");
backends.clear();
backends.addAll(orderedBackends);
+ backendMap.clear();
+
backendMap.putAll(backends.stream().collect(Collectors.groupingBy(Backend::getHost)));
nextBe = 0;
}
@@ -234,11 +246,6 @@ public class FederationBackendPolicy {
Collections.shuffle(splits, new Random(FIXED_SHUFFLE_SEED));
List<Split> remainingSplits;
-
- List<Backend> backends = new ArrayList<>();
- for (List<Backend> backendList : backendMap.values()) {
- backends.addAll(backendList);
- }
ResettableRandomizedIterator<Backend> randomCandidates = new
ResettableRandomizedIterator<>(backends);
// optimizedLocalScheduling enables prioritized assignment of splits
to local nodes when splits contain
@@ -330,9 +337,7 @@ public class FederationBackendPolicy {
}
List<Backend> allNodes = new ArrayList<>();
- for (List<Backend> backendList : backendMap.values()) {
- allNodes.addAll(backendList);
- }
+ allNodes.addAll(backends);
Collections.sort(allNodes, Comparator.comparing(Backend::getId));
if (allNodes.size() < 2) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
index 970d640c436..0f61ef0055c 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
@@ -94,9 +94,11 @@ import
org.apache.doris.nereids.trees.expressions.SessionVarGuardExpr;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.expressions.WindowFrame;
+import
org.apache.doris.nereids.trees.expressions.functions.NoneMovableFunction;
import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateParam;
import org.apache.doris.nereids.trees.expressions.functions.agg.AggregatePhase;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
import
org.apache.doris.nereids.trees.expressions.functions.scalar.GroupingScalarFunction;
import org.apache.doris.nereids.trees.plans.AbstractPlan;
import org.apache.doris.nereids.trees.plans.AggMode;
@@ -216,6 +218,11 @@ import org.apache.doris.planner.UnionNode;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.SessionVariable;
import org.apache.doris.statistics.StatisticConstants;
+import org.apache.doris.tablefunction.FileTableValuedFunction;
+import org.apache.doris.tablefunction.HdfsTableValuedFunction;
+import org.apache.doris.tablefunction.HttpTableValuedFunction;
+import org.apache.doris.tablefunction.LocalTableValuedFunction;
+import org.apache.doris.tablefunction.S3TableValuedFunction;
import org.apache.doris.tablefunction.TableValuedFunctionIf;
import org.apache.doris.thrift.TBinlogScanType;
import org.apache.doris.thrift.TExternalTableSinkWriterAssignment;
@@ -1148,6 +1155,10 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
TableValuedFunctionIf catalogFunction =
tvfRelation.getFunction().getCatalogFunction();
SessionVariable sv = ConnectContext.get().getSessionVariable();
ScanNode scanNode =
catalogFunction.getScanNode(context.nextPlanNodeId(), tupleDescriptor, sv);
+ if (context.getRelationPushAggOp(tvfRelation.getRelationId()) ==
TPushAggOp.COUNT) {
+ scanNode.setPushDownAggNoGrouping(TPushAggOp.COUNT);
+ scanNode.setPushDownCountSlotIds(ImmutableList.of());
+ }
scanNode.setEnableConditionCache(true);
scanNode.setDistributeExprLists(getDistributeExpr(tvfRelation));
scanNode.setNereidsId(tvfRelation.getId());
@@ -1193,6 +1204,10 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
return visitBucketedFusion(aggregate, context);
}
+ countPushDownFileTvf(aggregate,
ConnectContext.get().getSessionVariable()).ifPresent(tvf -> {
+ context.setRelationPushAggOp(tvf.getRelationId(),
TPushAggOp.COUNT);
+ context.setRelationPushCountArgumentExprIds(tvf.getRelationId(),
ImmutableList.of());
+ });
PlanFragment inputPlanFragment = aggregate.child(0).accept(this,
context);
List<List<Expr>> distributeExprLists =
getDistributeExprs(aggregate.child(0));
@@ -1259,7 +1274,8 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
// agg(update serialize) <- child, also set colocate = true
if (aggregate.getAggregateParam().aggMode.consumeAggregateBuffer
&& child instanceof PhysicalHashAggregate
- && !((PhysicalHashAggregate<Plan>)
child).getAggregateParam().aggMode.consumeAggregateBuffer
+ && !((PhysicalHashAggregate<Plan>)
child).getAggregateParam().aggMode
+ .consumeAggregateBuffer
&& inputPlanFragment.getPlanRoot() instanceof
AggregationNode) {
AggregationNode childAgg = (AggregationNode)
inputPlanFragment.getPlanRoot();
childAgg.setColocate(true);
@@ -1295,6 +1311,47 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
return inputPlanFragment;
}
+ /** Match only row-count aggregates over an unfiltered external file TVF.
*/
+ static Optional<PhysicalTVFRelation> countPushDownFileTvf(
+ PhysicalHashAggregate<? extends Plan> aggregate, SessionVariable
sessionVariable) {
+ Plan child = aggregate.child(0);
+ if (child instanceof PhysicalProject) {
+ PhysicalProject<?> project = (PhysicalProject<?>) child;
+ // COUNT readers synthesize column values, but retained
expressions still execute on those columns.
+ if (project.getProjects().stream()
+ .anyMatch(expression ->
expression.containsType(NoneMovableFunction.class))
+ ||
project.getMultiLayerProjects().stream().flatMap(List::stream)
+ .anyMatch(expression ->
expression.containsType(NoneMovableFunction.class))) {
+ return Optional.empty();
+ }
+ }
+ Plan tvfChild = child instanceof PhysicalProject && child.child(0)
instanceof PhysicalTVFRelation
+ ? child.child(0) : child;
+ Set<AggregateFunction> aggregateFunctions =
aggregate.getAggregateFunctions();
+ AggregateFunction aggregateFunction = aggregateFunctions.size() == 1
+ ? aggregateFunctions.iterator().next() : null;
+ if (tvfChild instanceof PhysicalTVFRelation
+ && isCountPushDownFileTvf(((PhysicalTVFRelation)
tvfChild).getFunction().getCatalogFunction())
+ && aggregate.getGroupByExpressions().isEmpty()
+ && aggregateFunction instanceof Count
+ && ((Count) aggregateFunction).isCountStar()
+ && !aggregateFunction.isDistinct()
+ &&
!aggregate.getAggregateParam().aggMode.consumeAggregateBuffer
+ && sessionVariable.enablePushDownNoGroupAgg()
+ && sessionVariable.isEnableCountPushDownForExternalTable()) {
+ return Optional.of((PhysicalTVFRelation) tvfChild);
+ }
+ return Optional.empty();
+ }
+
+ private static boolean isCountPushDownFileTvf(TableValuedFunctionIf
catalogFunction) {
+ return catalogFunction instanceof FileTableValuedFunction
+ || catalogFunction instanceof HdfsTableValuedFunction
+ || catalogFunction instanceof HttpTableValuedFunction
+ || catalogFunction instanceof LocalTableValuedFunction
+ || catalogFunction instanceof S3TableValuedFunction;
+ }
+
@Override
public PlanFragment visitPhysicalStorageLayerAggregate(
PhysicalStorageLayerAggregate storageLayerAggregate,
PlanTranslatorContext context) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/AbstractUnassignedScanJob.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/AbstractUnassignedScanJob.java
index 8c8a47e8444..1e747991af9 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/AbstractUnassignedScanJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/AbstractUnassignedScanJob.java
@@ -157,7 +157,7 @@ public abstract class AbstractUnassignedScanJob extends
AbstractUnassignedJob {
// now we should compute how many instances to process the data,
// for example: two instances
- int instanceNum = degreeOfParallelism(scanSourceMaxParallel,
useLocalShuffleToAddParallel);
+ int instanceNum = degreeOfParallelism(scanSource,
scanSourceMaxParallel, useLocalShuffleToAddParallel);
if (useLocalShuffleToAddParallel) {
assignLocalShuffleJobs(scanSource, instanceNum, instances,
context, worker);
@@ -265,6 +265,12 @@ public abstract class AbstractUnassignedScanJob extends
AbstractUnassignedJob {
}
}
+ /** Allow scan jobs to refine parallelism using the ranges assigned to
this worker. */
+ protected int degreeOfParallelism(ScanSource scanSource, int maxParallel,
+ boolean useLocalShuffleToAddParallel) {
+ return degreeOfParallelism(maxParallel, useLocalShuffleToAddParallel);
+ }
+
/**
* Compute the number of parallel instances for this fragment.
* The result is bounded by several constraints:
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanSingleRemoteTableJob.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanSingleRemoteTableJob.java
index e9dfc735880..23eb3dfe483 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanSingleRemoteTableJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanSingleRemoteTableJob.java
@@ -17,6 +17,7 @@
package org.apache.doris.nereids.trees.plans.distribute.worker.job;
+import org.apache.doris.datasource.tvf.source.TVFScanNode;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.trees.plans.distribute.DistributeContext;
import
org.apache.doris.nereids.trees.plans.distribute.worker.DistributedPlanWorker;
@@ -25,6 +26,8 @@ import
org.apache.doris.nereids.trees.plans.distribute.worker.ScanWorkerSelector
import org.apache.doris.planner.ExchangeNode;
import org.apache.doris.planner.PlanFragment;
import org.apache.doris.planner.ScanNode;
+import org.apache.doris.thrift.TFileScanRange;
+import org.apache.doris.thrift.TPushAggOp;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ListMultimap;
@@ -66,6 +69,31 @@ public class UnassignedScanSingleRemoteTableJob extends
AbstractUnassignedScanJo
);
}
+ /**
+ * Avoid extra local-shuffle instances for a file TVF row-count scan with
one
+ * actual split assigned to this worker. The scan still feeds the upper
COUNT
+ * aggregation; this shortcut only reduces fragment-instance overhead.
+ * Dynamic split sources and existing external-table COUNT paths retain
their
+ * original parallelism.
+ */
+ @Override
+ protected int degreeOfParallelism(ScanSource scanSource, int maxParallel,
+ boolean useLocalShuffleToAddParallel) {
+ ScanNode scanNode = scanNodes.get(0);
+ if (scanNode instanceof TVFScanNode &&
scanNode.getPushDownAggNoGroupingOp() == TPushAggOp.COUNT
+ && scanSource instanceof DefaultScanSource) {
+ ScanRanges scanRanges = ((DefaultScanSource)
scanSource).scanNodeToScanRanges.get(scanNode);
+ if (scanRanges.params.size() == 1) {
+ TFileScanRange fileScanRange =
scanRanges.params.get(0).getScanRange()
+ .getExtScanRange().getFileScanRange();
+ if (!fileScanRange.isSetSplitSource() &&
fileScanRange.getRangesSize() == 1) {
+ return 1;
+ }
+ }
+ }
+ return super.degreeOfParallelism(scanSource, maxParallel,
useLocalShuffleToAddParallel);
+ }
+
/**
* If all file scan ranges have been pruned and the assigned job list is
empty,
* create a single empty instance on a random worker so the fragment can
still
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/FileTvfCountPushDownTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/FileTvfCountPushDownTest.java
new file mode 100644
index 00000000000..e4cdd04fda1
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/FileTvfCountPushDownTest.java
@@ -0,0 +1,155 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.glue.translator;
+
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.GreaterThan;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateParam;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.AssertTrue;
+import
org.apache.doris.nereids.trees.expressions.functions.table.TableValuedFunction;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.StringLiteral;
+import org.apache.doris.nereids.trees.plans.AggMode;
+import org.apache.doris.nereids.trees.plans.AggPhase;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalHashAggregate;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalProject;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalTVFRelation;
+import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.tablefunction.CdcStreamTableValuedFunction;
+import org.apache.doris.tablefunction.FileTableValuedFunction;
+import org.apache.doris.tablefunction.GroupCommitTableValuedFunction;
+import org.apache.doris.tablefunction.HdfsTableValuedFunction;
+import org.apache.doris.tablefunction.HttpStreamTableValuedFunction;
+import org.apache.doris.tablefunction.HttpTableValuedFunction;
+import org.apache.doris.tablefunction.LocalTableValuedFunction;
+import org.apache.doris.tablefunction.S3TableValuedFunction;
+import org.apache.doris.tablefunction.TableValuedFunctionIf;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class FileTvfCountPushDownTest {
+ private PhysicalTVFRelation tvf(boolean file) {
+ return tvf(file ? FileTableValuedFunction.class :
TableValuedFunctionIf.class);
+ }
+
+ private PhysicalTVFRelation tvf(Class<? extends TableValuedFunctionIf>
functionClass) {
+ TableValuedFunction function = Mockito.mock(TableValuedFunction.class);
+
Mockito.when(function.getCatalogFunction()).thenReturn(Mockito.mock(functionClass));
+ PhysicalTVFRelation scan = Mockito.mock(PhysicalTVFRelation.class);
+ Mockito.when(scan.getFunction()).thenReturn(function);
+ return scan;
+ }
+
+ @SuppressWarnings("unchecked")
+ private PhysicalHashAggregate<Plan> aggregate(Plan child,
AggregateFunction function) {
+ PhysicalHashAggregate<Plan> aggregate =
Mockito.mock(PhysicalHashAggregate.class);
+ Mockito.when(aggregate.child(0)).thenReturn(child);
+
Mockito.when(aggregate.getGroupByExpressions()).thenReturn(ImmutableList.of());
+
Mockito.when(aggregate.getAggregateFunctions()).thenReturn(ImmutableSet.of(function));
+
Mockito.when(aggregate.getAggregateParam()).thenReturn(AggregateParam.LOCAL_BUFFER);
+ return aggregate;
+ }
+
+ @Test
+ public void countStarAndCountOneThroughProject() {
+ PhysicalTVFRelation scan = tvf(true);
+ PhysicalProject<?> project = Mockito.mock(PhysicalProject.class);
+ Mockito.doReturn(scan).when(project).child(0);
+ SessionVariable session = new SessionVariable();
+ Assertions.assertSame(scan,
PhysicalPlanTranslator.countPushDownFileTvf(
+ aggregate(scan, new Count()), session).get());
+ Assertions.assertSame(scan,
PhysicalPlanTranslator.countPushDownFileTvf(
+ aggregate(project, new Count(new IntegerLiteral(1))),
session).get());
+ }
+
+ @Test
+ public void rejectRetainedAssertionsInProjects() {
+ PhysicalTVFRelation scan = tvf(true);
+ SlotReference id = new SlotReference("id", IntegerType.INSTANCE);
+ Alias assertion = new Alias(new AssertTrue(new GreaterThan(id, new
IntegerLiteral(0)),
+ new StringLiteral("positive id")), "checked");
+ PhysicalProject<?> project = Mockito.mock(PhysicalProject.class);
+ Mockito.doReturn(scan).when(project).child(0);
+
Mockito.doReturn(ImmutableList.of(assertion)).when(project).getProjects();
+ SessionVariable session = new SessionVariable();
+ Assertions.assertFalse(PhysicalPlanTranslator.countPushDownFileTvf(
+ aggregate(project, new Count()), session).isPresent());
+
+ // The expression can survive in an intermediate layer even when the
final layer only forwards its slot.
+
Mockito.doReturn(ImmutableList.of(assertion.toSlot())).when(project).getProjects();
+ Mockito.doReturn(ImmutableList.of(ImmutableList.of(assertion),
ImmutableList.of(assertion.toSlot())))
+ .when(project).getMultiLayerProjects();
+ Assertions.assertFalse(PhysicalPlanTranslator.countPushDownFileTvf(
+ aggregate(project, new Count()), session).isPresent());
+ }
+
+ @Test
+ public void acceptAllSupportedFileTvfs() {
+ SessionVariable session = new SessionVariable();
+ for (Class<? extends TableValuedFunctionIf> functionClass :
ImmutableList.of(
+ FileTableValuedFunction.class,
+ HdfsTableValuedFunction.class,
+ HttpTableValuedFunction.class,
+ LocalTableValuedFunction.class,
+ S3TableValuedFunction.class)) {
+ Assertions.assertTrue(PhysicalPlanTranslator.countPushDownFileTvf(
+ aggregate(tvf(functionClass), new Count()),
session).isPresent());
+ }
+ }
+
+ @Test
+ public void rejectDistinctGroupedFilteredAndNonFilePlans() {
+ SessionVariable session = new SessionVariable();
+ Assertions.assertFalse(PhysicalPlanTranslator.countPushDownFileTvf(
+ aggregate(tvf(true), new Count(true, new IntegerLiteral(1))),
session).isPresent());
+ Assertions.assertFalse(PhysicalPlanTranslator.countPushDownFileTvf(
+ aggregate(tvf(false), new Count()), session).isPresent());
+ Assertions.assertFalse(PhysicalPlanTranslator.countPushDownFileTvf(
+ aggregate(Mockito.mock(Plan.class), new Count()),
session).isPresent());
+ PhysicalHashAggregate<Plan> grouped = aggregate(tvf(true), new
Count());
+
Mockito.when(grouped.getGroupByExpressions()).thenReturn(ImmutableList.of(new
IntegerLiteral(1)));
+
Assertions.assertFalse(PhysicalPlanTranslator.countPushDownFileTvf(grouped,
session).isPresent());
+ Assertions.assertFalse(PhysicalPlanTranslator.countPushDownFileTvf(
+ aggregate(tvf(CdcStreamTableValuedFunction.class), new
Count()), session).isPresent());
+ Assertions.assertFalse(PhysicalPlanTranslator.countPushDownFileTvf(
+ aggregate(tvf(GroupCommitTableValuedFunction.class), new
Count()), session).isPresent());
+ Assertions.assertFalse(PhysicalPlanTranslator.countPushDownFileTvf(
+ aggregate(tvf(HttpStreamTableValuedFunction.class), new
Count()), session).isPresent());
+ }
+
+ @Test
+ public void rejectMergeStageAndDisabledPushDown() {
+ SessionVariable session = new SessionVariable();
+ PhysicalHashAggregate<Plan> aggregate = aggregate(tvf(true), new
Count());
+ Mockito.when(aggregate.getAggregateParam()).thenReturn(
+ new AggregateParam(AggPhase.GLOBAL, AggMode.BUFFER_TO_RESULT));
+
Assertions.assertFalse(PhysicalPlanTranslator.countPushDownFileTvf(aggregate,
session).isPresent());
+
Mockito.when(aggregate.getAggregateParam()).thenReturn(AggregateParam.LOCAL_BUFFER);
+ session.enablePushDownNoGroupAgg = false;
+
Assertions.assertFalse(PhysicalPlanTranslator.countPushDownFileTvf(aggregate,
session).isPresent());
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanSingleRemoteTableJobTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanSingleRemoteTableJobTest.java
new file mode 100644
index 00000000000..40734d9af8b
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedScanSingleRemoteTableJobTest.java
@@ -0,0 +1,162 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.trees.plans.distribute.worker.job;
+
+import org.apache.doris.datasource.tvf.source.TVFScanNode;
+import org.apache.doris.nereids.StatementContext;
+import org.apache.doris.nereids.trees.plans.distribute.DistributeContext;
+import
org.apache.doris.nereids.trees.plans.distribute.worker.DistributedPlanWorker;
+import
org.apache.doris.nereids.trees.plans.distribute.worker.DistributedPlanWorkerManager;
+import
org.apache.doris.nereids.trees.plans.distribute.worker.ScanWorkerSelector;
+import org.apache.doris.planner.DataPartition;
+import org.apache.doris.planner.PlanFragment;
+import org.apache.doris.planner.ScanNode;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.OriginStatement;
+import org.apache.doris.thrift.TExternalScanRange;
+import org.apache.doris.thrift.TFileRangeDesc;
+import org.apache.doris.thrift.TFileScanRange;
+import org.apache.doris.thrift.TPushAggOp;
+import org.apache.doris.thrift.TScanRange;
+import org.apache.doris.thrift.TScanRangeParams;
+import org.apache.doris.thrift.TSplitSource;
+import org.apache.doris.thrift.TUniqueId;
+
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.List;
+
+public class UnassignedScanSingleRemoteTableJobTest {
+
+ @Test
+ public void testOneAssignedRangeCountPushdownUsesOneInstance() {
+ UnassignedScanSingleRemoteTableJob job = createJob(TPushAggOp.COUNT);
+
+ Assertions.assertEquals(1, job.degreeOfParallelism(source(job, 1,
false), 1, true));
+ }
+
+ @Test
+ public void
testCountWithMultiplePhysicalSplitsKeepsLocalShuffleParallelism() {
+ UnassignedScanSingleRemoteTableJob job = createJob(TPushAggOp.COUNT);
+
+ Assertions.assertEquals(8, job.degreeOfParallelism(source(job, 2,
false), 1, true));
+ }
+
+ @Test
+ public void testNonCountKeepsLocalShuffleParallelism() {
+ UnassignedScanSingleRemoteTableJob job = createJob(TPushAggOp.NONE);
+
+ Assertions.assertEquals(8, job.degreeOfParallelism(source(job, 1,
false), 1, true));
+ }
+
+ @Test
+ public void
testCountWithMultipleRangeEntriesKeepsLocalShuffleParallelism() {
+ UnassignedScanSingleRemoteTableJob job = createJob(TPushAggOp.COUNT);
+ DefaultScanSource source = source(job, 1, false);
+ ScanRanges ranges =
source.scanNodeToScanRanges.get(job.scanNodes.get(0));
+ DefaultScanSource multipleRanges = new
DefaultScanSource(ImmutableMap.of(job.scanNodes.get(0),
+ new ScanRanges(ImmutableList.of(ranges.params.get(0),
ranges.params.get(0)),
+ ImmutableList.of(1L, 1L))));
+ Assertions.assertEquals(8, job.degreeOfParallelism(multipleRanges, 2,
true));
+ }
+
+ @Test
+ public void testDynamicSplitSourceKeepsLocalShuffleParallelism() {
+ UnassignedScanSingleRemoteTableJob job = createJob(TPushAggOp.COUNT);
+ Assertions.assertEquals(8, job.degreeOfParallelism(source(job, 0,
true), 1, true));
+ }
+
+ @Test
+ public void testExternalTableCountKeepsLocalShuffleParallelism() {
+ UnassignedScanSingleRemoteTableJob job = createJob(TPushAggOp.COUNT,
ScanNode.class);
+ Assertions.assertEquals(8, job.degreeOfParallelism(source(job, 1,
false), 1, true));
+ }
+
+ @Test
+ public void testAssignedInstancesPreserveDynamicSource() {
+ UnassignedScanSingleRemoteTableJob job = createJob(TPushAggOp.COUNT);
+ DefaultScanSource source = source(job, 0, true);
+
Mockito.when(job.fragment.useSerialSource(Mockito.any())).thenReturn(true);
+ DistributedPlanWorker worker =
Mockito.mock(DistributedPlanWorker.class);
+ DistributeContext context = new
DistributeContext(Mockito.mock(DistributedPlanWorkerManager.class), false);
+
+ List<AssignedJob> instances = job.insideMachineParallelization(
+ ImmutableMap.of(worker, new UninstancedScanSource(source)),
ArrayListMultimap.create(), context);
+
+ Assertions.assertEquals(8, instances.size());
+ Assertions.assertSame(source, instances.get(0).getScanSource());
+ for (int i = 0; i < instances.size(); i++) {
+ Assertions.assertInstanceOf(LocalShuffleAssignedJob.class,
instances.get(i));
+ Assertions.assertEquals(0, ((LocalShuffleAssignedJob)
instances.get(i)).shareScanId);
+ if (i > 0) {
+
Assertions.assertTrue(instances.get(i).getScanSource().isEmpty());
+ }
+ }
+ }
+
+ private static DefaultScanSource source(UnassignedScanSingleRemoteTableJob
job,
+ int splitCount, boolean dynamic) {
+ TFileScanRange fileRange = new TFileScanRange();
+ for (int i = 0; i < splitCount; i++) {
+ fileRange.addToRanges(new TFileRangeDesc());
+ }
+ if (dynamic) {
+ fileRange.setSplitSource(new TSplitSource());
+ }
+ TScanRange range = new TScanRange();
+ range.setExtScanRange(new
TExternalScanRange().setFileScanRange(fileRange));
+ TScanRangeParams params = new TScanRangeParams().setScanRange(range);
+ return new DefaultScanSource(ImmutableMap.of(job.scanNodes.get(0),
+ new ScanRanges(ImmutableList.of(params),
ImmutableList.of(1L))));
+ }
+
+ private static UnassignedScanSingleRemoteTableJob createJob(TPushAggOp
aggOp) {
+ return createJob(aggOp, TVFScanNode.class);
+ }
+
+ private static UnassignedScanSingleRemoteTableJob createJob(TPushAggOp
aggOp,
+ Class<? extends ScanNode> scanNodeClass) {
+ ConnectContext connectContext = new ConnectContext();
+ connectContext.setThreadLocalInfo();
+ connectContext.setQueryId(new TUniqueId(10, 10));
+ StatementContext statementContext = new StatementContext(
+ connectContext, new OriginStatement("select count(*) from
files", 0));
+ connectContext.setStatementContext(statementContext);
+
+ ScanNode scanNode = Mockito.mock(scanNodeClass);
+ Mockito.when(scanNode.getPushDownAggNoGroupingOp()).thenReturn(aggOp);
+ // The global scan range count can differ from the ranges assigned to
this worker.
+ Mockito.when(scanNode.getScanRangeNum()).thenReturn(1);
+
+ PlanFragment fragment = Mockito.mock(PlanFragment.class);
+
Mockito.when(fragment.getDataPartition()).thenReturn(DataPartition.RANDOM);
+ Mockito.when(fragment.getParallelExecNum()).thenReturn(8);
+
+ return new UnassignedScanSingleRemoteTableJob(
+ statementContext,
+ fragment,
+ scanNode,
+ ArrayListMultimap.create(),
+ Mockito.mock(ScanWorkerSelector.class));
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/planner/FederationBackendPolicyTest.java
b/fe/fe-core/src/test/java/org/apache/doris/planner/FederationBackendPolicyTest.java
index d28dc0ffe59..de51e607f47 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/planner/FederationBackendPolicyTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/planner/FederationBackendPolicyTest.java
@@ -67,6 +67,78 @@ public class FederationBackendPolicyTest {
mockedEnvStatic.close();
}
+ @Test
+ public void testRoundRobinUsesCandidateOrderAcrossBatches() throws
UserException {
+ SystemInfoService service = new SystemInfoService();
+ Backend first = new Backend(10002L, "172.30.0.100", 9050);
+ Backend second = new Backend(10003L, "172.30.0.106", 9050);
+ Backend third = new Backend(10004L, "172.30.0.118", 9050);
+ for (Backend backend : List.of(first, second, third)) {
+ backend.setAlive(true);
+ service.addBackend(backend);
+ }
+ mockedEnvStatic.when(Env::getCurrentSystemInfo).thenReturn(service);
+ Mockito.when(env.getComputeGroupMgr()).thenReturn(new
ComputeGroupMgr(service));
+
+ // Test different query-local candidate orders. A Host map has one
stable iteration order,
+ // so it cannot satisfy both first-split expectations.
+ for (List<Backend> order : List.of(List.of(first, second, third),
List.of(third, first, second))) {
+ FederationBackendPolicy policy = new FederationBackendPolicy();
+ policy.init();
+ policy.replaceBackendOrder(order);
+ policy.setEnableSplitsRedistribution(false);
+ for (int i = 0; i < 4; i++) {
+ FileSplit split = new
FileSplit(LocationPath.of("hdfs://example/test/part-" + i),
+ 0, 1, 1, 0, null, Collections.emptyList());
+ Multimap<Backend, Split> assignment =
policy.computeScanRangeAssignment(
+ new ArrayList<>(List.of(split)));
+ Assertions.assertEquals(order.get(i % order.size()),
assignment.keySet().iterator().next());
+ }
+
+ policy.replaceBackendOrder(List.of(first, second));
+ FileSplit excludedLocalSplit = new
FileSplit(LocationPath.of("hdfs://example/excluded-local-part"),
+ 0, 1, 1, 0, new String[]{third.getHost()},
Collections.emptyList());
+ Multimap<Backend, Split> assignment =
policy.computeScanRangeAssignment(
+ new ArrayList<>(List.of(excludedLocalSplit)));
+ Assertions.assertTrue(assignment.keySet().stream()
+ .allMatch(backend -> backend.equals(first) ||
backend.equals(second)));
+ }
+ }
+
+ @Test
+ public void testRoundRobinPreservesOrderForBackendsOnSameHost() throws
UserException {
+ SystemInfoService service = new SystemInfoService();
+ Backend first = new Backend(11002L, "172.30.0.100", 9050);
+ Backend second = new Backend(11003L, "172.30.0.100", 9051);
+ Backend third = new Backend(11004L, "172.30.0.100", 9052);
+ for (Backend backend : List.of(first, second, third)) {
+ backend.setAlive(true);
+ service.addBackend(backend);
+ }
+ mockedEnvStatic.when(Env::getCurrentSystemInfo).thenReturn(service);
+ Mockito.when(env.getComputeGroupMgr()).thenReturn(new
ComputeGroupMgr(service));
+
+ FederationBackendPolicy policy = new FederationBackendPolicy();
+ policy.init();
+ policy.replaceBackendOrder(List.of(second, third, first));
+ policy.setEnableSplitsRedistribution(false);
+ for (int i = 0; i < 3; i++) {
+ FileSplit split = new
FileSplit(LocationPath.of("hdfs://example/same-host-part-" + i),
+ 0, 1, 1, 0, null, Collections.emptyList());
+ Multimap<Backend, Split> assignment =
policy.computeScanRangeAssignment(
+ new ArrayList<>(List.of(split)));
+ Assertions.assertEquals(List.of(second, third, first).get(i),
+ assignment.keySet().iterator().next());
+ }
+ Assertions.assertDoesNotThrow(() ->
policy.replaceBackendOrder(List.of(first, second)));
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () -> policy.replaceBackendOrder(List.of(first, first,
third)));
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () -> policy.replaceBackendOrder(List.of(new Backend(11005L,
"172.30.0.101", 9050))));
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () -> policy.replaceBackendOrder(Collections.emptyList()));
+ }
+
@Test
public void testRemoteSplits() throws UserException {
SystemInfoService service = new SystemInfoService();
@@ -102,6 +174,9 @@ public class FederationBackendPolicyTest {
Multimap<Backend, Split> assignment =
policy.computeScanRangeAssignment(splits);
+ Assertions.assertEquals(splits.size(), assignment.values().size());
+ Assertions.assertTrue(assignment.keySet().stream()
+ .allMatch(backend -> backend.equals(backend1) ||
backend.equals(backend2) || backend.equals(backend3)));
for (Backend backend : assignment.keySet()) {
Collection<Split> assignedSplits = assignment.get(backend);
long scanBytes = 0L;
@@ -221,6 +296,9 @@ public class FederationBackendPolicyTest {
Multimap<Backend, Split> assignment =
policy.computeScanRangeAssignment(splits);
+ Assertions.assertEquals(splits.size(), assignment.values().size());
+ Assertions.assertTrue(assignment.keySet().stream()
+ .allMatch(backend -> backend.equals(backend1) ||
backend.equals(backend2) || backend.equals(backend3)));
int maxAssignedSplitNum = Integer.MIN_VALUE;
int minAssignedSplitNum = Integer.MAX_VALUE;
for (Backend backend : assignment.keySet()) {
@@ -242,6 +320,49 @@ public class FederationBackendPolicyTest {
}
+ @Test
+ public void testConsistentHashSingleSplitTieIsStableAcrossPolicies()
throws UserException {
+ SystemInfoService service = new SystemInfoService();
+ Backend backend1 = new Backend(10002L, "172.30.0.100", 9050);
+ Backend backend2 = new Backend(10003L, "172.30.0.106", 9050);
+ Backend backend3 = new Backend(10004L, "172.30.0.118", 9050);
+ for (Backend backend : List.of(backend1, backend2, backend3)) {
+ backend.setAlive(true);
+ service.addBackend(backend);
+ }
+ ComputeGroupMgr cgmgr = new ComputeGroupMgr(service);
+ mockedEnvStatic.when(Env::getCurrentSystemInfo).thenReturn(service);
+ Mockito.when(env.getComputeGroupMgr()).thenReturn(cgmgr);
+
+ int originalCandidateNum =
Config.split_assigner_min_consistent_hash_candidate_num;
+ Config.split_assigner_min_consistent_hash_candidate_num = 2;
+ try {
+ Backend assignedBackend = null;
+ for (int i = 0; i < 10; i++) {
+ FederationBackendPolicy policy = new
FederationBackendPolicy(NodeSelectionStrategy.CONSISTENT_HASHING);
+ policy.init();
+ policy.setEnableSplitsRedistribution(false);
+ FileSplit split = new
FileSplit(LocationPath.of("hdfs://example/cache-hotspot.csv"),
+ 0, 1024, 1024, 0, null, Collections.emptyList());
+ Multimap<Backend, Split> assignment =
policy.computeScanRangeAssignment(
+ new ArrayList<>(List.of(split)));
+ Backend currentBackend = assignment.keySet().iterator().next();
+ if (assignedBackend == null) {
+ assignedBackend = currentBackend;
+ } else {
+ Assertions.assertEquals(assignedBackend, currentBackend);
+ }
+ }
+ FederationBackendPolicy consistentHashPolicy =
+ new
FederationBackendPolicy(NodeSelectionStrategy.CONSISTENT_HASHING);
+ consistentHashPolicy.init();
+ Assertions.assertThrows(IllegalStateException.class,
+ () ->
consistentHashPolicy.replaceBackendOrder(List.of(backend1, backend2)));
+ } finally {
+ Config.split_assigner_min_consistent_hash_candidate_num =
originalCandidateNum;
+ }
+ }
+
public static void sortSplits(List<Split> splits) {
splits.sort((split1, split2) -> {
int pathComparison =
split1.getPathString().compareTo(split2.getPathString());
diff --git
a/regression-test/data/external_table_p0/tvf/test_file_tvf_count_pushdown.out
b/regression-test/data/external_table_p0/tvf/test_file_tvf_count_pushdown.out
new file mode 100644
index 00000000000..4ec6a63659b
--- /dev/null
+++
b/regression-test/data/external_table_p0/tvf/test_file_tvf_count_pushdown.out
@@ -0,0 +1,428 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !v2_false_push_false_csv_star --
+4
+
+-- !v2_false_push_false_csv_one --
+4
+
+-- !v2_false_push_false_csv_nullable --
+3
+
+-- !v2_false_push_false_csv_distinct --
+2
+
+-- !v2_false_push_false_csv_filter --
+2
+
+-- !v2_false_push_false_csv_group --
+\N 1
+10 2
+20 1
+
+-- !v2_false_push_false_csv_empty --
+0
+
+-- !v2_false_push_false_csv_multiple --
+8
+
+-- !v2_false_push_false_json_star --
+4
+
+-- !v2_false_push_false_json_one --
+4
+
+-- !v2_false_push_false_json_nullable --
+3
+
+-- !v2_false_push_false_json_distinct --
+2
+
+-- !v2_false_push_false_json_filter --
+2
+
+-- !v2_false_push_false_json_group --
+\N 1
+10 2
+20 1
+
+-- !v2_false_push_false_json_multiple --
+8
+
+-- !v2_false_push_false_parquet_star --
+4
+
+-- !v2_false_push_false_parquet_one --
+4
+
+-- !v2_false_push_false_parquet_nullable --
+3
+
+-- !v2_false_push_false_parquet_distinct --
+2
+
+-- !v2_false_push_false_parquet_filter --
+2
+
+-- !v2_false_push_false_parquet_group --
+\N 1
+10 2
+20 1
+
+-- !v2_false_push_false_parquet_empty --
+0
+
+-- !v2_false_push_false_parquet_multiple --
+8
+
+-- !v2_false_push_false_orc_star --
+4
+
+-- !v2_false_push_false_orc_one --
+4
+
+-- !v2_false_push_false_orc_nullable --
+3
+
+-- !v2_false_push_false_orc_distinct --
+2
+
+-- !v2_false_push_false_orc_filter --
+2
+
+-- !v2_false_push_false_orc_group --
+\N 1
+10 2
+20 1
+
+-- !v2_false_push_false_orc_empty --
+0
+
+-- !v2_false_push_false_orc_multiple --
+8
+
+-- !v2_false_push_false_csv_header --
+4
+
+-- !v2_false_push_false_csv_header_nullable --
+3
+
+-- !v2_false_push_true_csv_star --
+4
+
+-- !v2_false_push_true_csv_one --
+4
+
+-- !v2_false_push_true_csv_nullable --
+3
+
+-- !v2_false_push_true_csv_distinct --
+2
+
+-- !v2_false_push_true_csv_filter --
+2
+
+-- !v2_false_push_true_csv_group --
+\N 1
+10 2
+20 1
+
+-- !v2_false_push_true_csv_empty --
+0
+
+-- !v2_false_push_true_csv_multiple --
+8
+
+-- !v2_false_push_true_json_star --
+4
+
+-- !v2_false_push_true_json_one --
+4
+
+-- !v2_false_push_true_json_nullable --
+3
+
+-- !v2_false_push_true_json_distinct --
+2
+
+-- !v2_false_push_true_json_filter --
+2
+
+-- !v2_false_push_true_json_group --
+\N 1
+10 2
+20 1
+
+-- !v2_false_push_true_json_multiple --
+8
+
+-- !v2_false_push_true_parquet_star --
+4
+
+-- !v2_false_push_true_parquet_one --
+4
+
+-- !v2_false_push_true_parquet_nullable --
+3
+
+-- !v2_false_push_true_parquet_distinct --
+2
+
+-- !v2_false_push_true_parquet_filter --
+2
+
+-- !v2_false_push_true_parquet_group --
+\N 1
+10 2
+20 1
+
+-- !v2_false_push_true_parquet_empty --
+0
+
+-- !v2_false_push_true_parquet_multiple --
+8
+
+-- !v2_false_push_true_orc_star --
+4
+
+-- !v2_false_push_true_orc_one --
+4
+
+-- !v2_false_push_true_orc_nullable --
+3
+
+-- !v2_false_push_true_orc_distinct --
+2
+
+-- !v2_false_push_true_orc_filter --
+2
+
+-- !v2_false_push_true_orc_group --
+\N 1
+10 2
+20 1
+
+-- !v2_false_push_true_orc_empty --
+0
+
+-- !v2_false_push_true_orc_multiple --
+8
+
+-- !v2_false_push_true_csv_header --
+4
+
+-- !v2_false_push_true_csv_header_nullable --
+3
+
+-- !v2_true_push_false_csv_star --
+4
+
+-- !v2_true_push_false_csv_one --
+4
+
+-- !v2_true_push_false_csv_nullable --
+3
+
+-- !v2_true_push_false_csv_distinct --
+2
+
+-- !v2_true_push_false_csv_filter --
+2
+
+-- !v2_true_push_false_csv_group --
+\N 1
+10 2
+20 1
+
+-- !v2_true_push_false_csv_empty --
+0
+
+-- !v2_true_push_false_csv_multiple --
+8
+
+-- !v2_true_push_false_json_star --
+4
+
+-- !v2_true_push_false_json_one --
+4
+
+-- !v2_true_push_false_json_nullable --
+3
+
+-- !v2_true_push_false_json_distinct --
+2
+
+-- !v2_true_push_false_json_filter --
+2
+
+-- !v2_true_push_false_json_group --
+\N 1
+10 2
+20 1
+
+-- !v2_true_push_false_json_multiple --
+8
+
+-- !v2_true_push_false_parquet_star --
+4
+
+-- !v2_true_push_false_parquet_one --
+4
+
+-- !v2_true_push_false_parquet_nullable --
+3
+
+-- !v2_true_push_false_parquet_distinct --
+2
+
+-- !v2_true_push_false_parquet_filter --
+2
+
+-- !v2_true_push_false_parquet_group --
+\N 1
+10 2
+20 1
+
+-- !v2_true_push_false_parquet_empty --
+0
+
+-- !v2_true_push_false_parquet_multiple --
+8
+
+-- !v2_true_push_false_orc_star --
+4
+
+-- !v2_true_push_false_orc_one --
+4
+
+-- !v2_true_push_false_orc_nullable --
+3
+
+-- !v2_true_push_false_orc_distinct --
+2
+
+-- !v2_true_push_false_orc_filter --
+2
+
+-- !v2_true_push_false_orc_group --
+\N 1
+10 2
+20 1
+
+-- !v2_true_push_false_orc_empty --
+0
+
+-- !v2_true_push_false_orc_multiple --
+8
+
+-- !v2_true_push_false_csv_header --
+4
+
+-- !v2_true_push_false_csv_header_nullable --
+3
+
+-- !v2_true_push_true_csv_star --
+4
+
+-- !v2_true_push_true_csv_one --
+4
+
+-- !v2_true_push_true_csv_nullable --
+3
+
+-- !v2_true_push_true_csv_distinct --
+2
+
+-- !v2_true_push_true_csv_filter --
+2
+
+-- !v2_true_push_true_csv_group --
+\N 1
+10 2
+20 1
+
+-- !v2_true_push_true_csv_empty --
+0
+
+-- !v2_true_push_true_csv_multiple --
+8
+
+-- !v2_true_push_true_json_star --
+4
+
+-- !v2_true_push_true_json_one --
+4
+
+-- !v2_true_push_true_json_nullable --
+3
+
+-- !v2_true_push_true_json_distinct --
+2
+
+-- !v2_true_push_true_json_filter --
+2
+
+-- !v2_true_push_true_json_group --
+\N 1
+10 2
+20 1
+
+-- !v2_true_push_true_json_multiple --
+8
+
+-- !v2_true_push_true_parquet_star --
+4
+
+-- !v2_true_push_true_parquet_one --
+4
+
+-- !v2_true_push_true_parquet_nullable --
+3
+
+-- !v2_true_push_true_parquet_distinct --
+2
+
+-- !v2_true_push_true_parquet_filter --
+2
+
+-- !v2_true_push_true_parquet_group --
+\N 1
+10 2
+20 1
+
+-- !v2_true_push_true_parquet_empty --
+0
+
+-- !v2_true_push_true_parquet_multiple --
+8
+
+-- !v2_true_push_true_orc_star --
+4
+
+-- !v2_true_push_true_orc_one --
+4
+
+-- !v2_true_push_true_orc_nullable --
+3
+
+-- !v2_true_push_true_orc_distinct --
+2
+
+-- !v2_true_push_true_orc_filter --
+2
+
+-- !v2_true_push_true_orc_group --
+\N 1
+10 2
+20 1
+
+-- !v2_true_push_true_orc_empty --
+0
+
+-- !v2_true_push_true_orc_multiple --
+8
+
+-- !v2_true_push_true_csv_header --
+4
+
+-- !v2_true_push_true_csv_header_nullable --
+3
diff --git
a/regression-test/data/external_table_p0/tvf/test_s3_tvf_compression.out
b/regression-test/data/external_table_p0/tvf/test_s3_tvf_compression.out
index 1308b7ffef2..b8816129916 100644
--- a/regression-test/data/external_table_p0/tvf/test_s3_tvf_compression.out
+++ b/regression-test/data/external_table_p0/tvf/test_s3_tvf_compression.out
@@ -43,6 +43,12 @@
19 2024-05-27
20 2023-12-18
+-- !gz_count_star --
+2000
+
+-- !gz_count_one --
+2000
+
-- !bz2_1 --
1 694832 buHDwfGeNHfpRFdNaogneddi 2024-02-09
4.899588807225554
10 218729 goZsLvvWFOIjlzSAitC 2024-06-10 4.137732740231178
diff --git
a/regression-test/data/external_table_p0/tvf/tvf_count_data_empty.csv
b/regression-test/data/external_table_p0/tvf/tvf_count_data_empty.csv
new file mode 100644
index 00000000000..e69de29bb2d
diff --git
a/regression-test/data/external_table_p0/tvf/tvf_count_data_empty.json
b/regression-test/data/external_table_p0/tvf/tvf_count_data_empty.json
new file mode 100644
index 00000000000..e69de29bb2d
diff --git
a/regression-test/data/external_table_p0/tvf/tvf_count_data_empty.orc
b/regression-test/data/external_table_p0/tvf/tvf_count_data_empty.orc
new file mode 100644
index 00000000000..ea26cb5446c
Binary files /dev/null and
b/regression-test/data/external_table_p0/tvf/tvf_count_data_empty.orc differ
diff --git
a/regression-test/data/external_table_p0/tvf/tvf_count_data_empty.parquet
b/regression-test/data/external_table_p0/tvf/tvf_count_data_empty.parquet
new file mode 100644
index 00000000000..9c1db5b0db5
Binary files /dev/null and
b/regression-test/data/external_table_p0/tvf/tvf_count_data_empty.parquet differ
diff --git a/regression-test/data/external_table_p0/tvf/tvf_count_data_more.csv
b/regression-test/data/external_table_p0/tvf/tvf_count_data_more.csv
new file mode 100644
index 00000000000..c602f80b56d
--- /dev/null
+++ b/regression-test/data/external_table_p0/tvf/tvf_count_data_more.csv
@@ -0,0 +1,4 @@
+1,10
+2,\N
+3,10
+4,20
diff --git
a/regression-test/data/external_table_p0/tvf/tvf_count_data_more.json
b/regression-test/data/external_table_p0/tvf/tvf_count_data_more.json
new file mode 100644
index 00000000000..2ea20d84a43
--- /dev/null
+++ b/regression-test/data/external_table_p0/tvf/tvf_count_data_more.json
@@ -0,0 +1,4 @@
+{"id": 1, "value": 10}
+{"id": 2, "value": null}
+{"id": 3, "value": 10}
+{"id": 4, "value": 20}
diff --git a/regression-test/data/external_table_p0/tvf/tvf_count_data_more.orc
b/regression-test/data/external_table_p0/tvf/tvf_count_data_more.orc
new file mode 100644
index 00000000000..950dae1cd75
Binary files /dev/null and
b/regression-test/data/external_table_p0/tvf/tvf_count_data_more.orc differ
diff --git
a/regression-test/data/external_table_p0/tvf/tvf_count_data_more.parquet
b/regression-test/data/external_table_p0/tvf/tvf_count_data_more.parquet
new file mode 100644
index 00000000000..1ec84d6b9a6
Binary files /dev/null and
b/regression-test/data/external_table_p0/tvf/tvf_count_data_more.parquet differ
diff --git a/regression-test/data/external_table_p0/tvf/tvf_count_data_rows.csv
b/regression-test/data/external_table_p0/tvf/tvf_count_data_rows.csv
new file mode 100644
index 00000000000..c602f80b56d
--- /dev/null
+++ b/regression-test/data/external_table_p0/tvf/tvf_count_data_rows.csv
@@ -0,0 +1,4 @@
+1,10
+2,\N
+3,10
+4,20
diff --git
a/regression-test/data/external_table_p0/tvf/tvf_count_data_rows.json
b/regression-test/data/external_table_p0/tvf/tvf_count_data_rows.json
new file mode 100644
index 00000000000..2ea20d84a43
--- /dev/null
+++ b/regression-test/data/external_table_p0/tvf/tvf_count_data_rows.json
@@ -0,0 +1,4 @@
+{"id": 1, "value": 10}
+{"id": 2, "value": null}
+{"id": 3, "value": 10}
+{"id": 4, "value": 20}
diff --git a/regression-test/data/external_table_p0/tvf/tvf_count_data_rows.orc
b/regression-test/data/external_table_p0/tvf/tvf_count_data_rows.orc
new file mode 100644
index 00000000000..950dae1cd75
Binary files /dev/null and
b/regression-test/data/external_table_p0/tvf/tvf_count_data_rows.orc differ
diff --git
a/regression-test/data/external_table_p0/tvf/tvf_count_data_rows.parquet
b/regression-test/data/external_table_p0/tvf/tvf_count_data_rows.parquet
new file mode 100644
index 00000000000..1ec84d6b9a6
Binary files /dev/null and
b/regression-test/data/external_table_p0/tvf/tvf_count_data_rows.parquet differ
diff --git a/regression-test/data/external_table_p0/tvf/tvf_count_header.csv
b/regression-test/data/external_table_p0/tvf/tvf_count_header.csv
new file mode 100644
index 00000000000..dab502f3fb0
--- /dev/null
+++ b/regression-test/data/external_table_p0/tvf/tvf_count_header.csv
@@ -0,0 +1,6 @@
+id,value
+1,10
+
+2,\N
+3,10
+4,20
\ No newline at end of file
diff --git
a/regression-test/suites/cloud_p0/test_read_cluster_var_property.groovy
b/regression-test/suites/cloud_p0/test_read_cluster_var_property.groovy
index df767e4409a..6a254cc0369 100644
--- a/regression-test/suites/cloud_p0/test_read_cluster_var_property.groovy
+++ b/regression-test/suites/cloud_p0/test_read_cluster_var_property.groovy
@@ -42,6 +42,8 @@ suite('test_read_cluster_var_property') {
}
connect(userName, "${pwd}", context.config.jdbcUrl) {
+ // Each metric assertion must execute a scan on the BE rather than
reuse a cached SQL result.
+ sql "set enable_sql_cache=false;"
// test non-mow table
try {
def tableName = "test_read_cluster_var_property"
@@ -214,4 +216,4 @@ suite('test_read_cluster_var_property') {
sql "set property for '${userName}'
query_freshness_tolerance_ms=-1;"
}
}
-}
\ No newline at end of file
+}
diff --git
a/regression-test/suites/external_table_p0/tvf/test_file_tvf_count_pushdown.groovy
b/regression-test/suites/external_table_p0/tvf/test_file_tvf_count_pushdown.groovy
new file mode 100644
index 00000000000..d2545a7ad95
--- /dev/null
+++
b/regression-test/suites/external_table_p0/tvf/test_file_tvf_count_pushdown.groovy
@@ -0,0 +1,111 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_file_tvf_count_pushdown", "p0,external") {
+ def backends = sql "show backends"
+ def backendId = backends[0][0]
+ def dataPath = context.config.dataPath + "/external_table_p0/tvf"
+ def predeployedPath =
context.config.otherConfigs.get("fileTvfCountFixturePath")
+ def remotePath = predeployedPath ?: "/tmp/file_tvf_count_pushdown"
+ def fixtures = ["tvf_count_header.csv"]
+ for (def format : ["csv", "json", "parquet", "orc"]) {
+ fixtures.add("tvf_count_data_rows.${format}")
+ fixtures.add("tvf_count_data_empty.${format}")
+ fixtures.add("tvf_count_data_more.${format}")
+ }
+ if (predeployedPath == null) {
+ mkdirRemotePathOnAllBE("root", remotePath)
+ for (def backend : backends) {
+ for (def fixture : fixtures) {
+ scpFiles("root", backend[1], "${dataPath}/${fixture}",
remotePath, false)
+ }
+ }
+ }
+
+ // The same snapshots exercise both the record-reading and row-count paths.
+ // Each nonempty fixture contains four rows, with one NULL and a duplicate
value.
+ for (def scannerV2 : [false, true]) {
+ sql "set enable_file_scanner_v2 = ${scannerV2}"
+ for (def pushdown : [false, true]) {
+ sql "set enable_push_down_no_group_agg = ${pushdown}"
+ def mode = "v2_${scannerV2}_push_${pushdown}"
+ for (def format : ["csv", "json", "parquet", "orc"]) {
+ def options = ""
+ if (format == "csv") {
+ options = ', "column_separator" = ",", "csv_schema" =
"id:int;value:int"'
+ } else if (format == "json") {
+ options = ', "read_json_by_line" = "true"'
+ }
+ def table = """local("file_path" =
"${remotePath}/tvf_count_data_rows.${format}",
+ "backend_id" = "${backendId}", "format" = "${format}"
${options})"""
+ explain {
+ sql "SELECT COUNT(*) FROM ${table}"
+ contains "pushdown agg=${pushdown ? 'COUNT' : 'NONE'}"
+ }
+ for (def query : ["SELECT COUNT(value) FROM ${table}",
+ "SELECT COUNT(DISTINCT value) FROM ${table}",
+ "SELECT COUNT(*) FROM ${table} WHERE id > 2",
+ "SELECT value, COUNT(*) FROM ${table} GROUP BY
value"]) {
+ explain {
+ sql query
+ contains "pushdown agg=NONE"
+ }
+ }
+ quickTest("${mode}_${format}_star", "SELECT COUNT(*) FROM
${table}")
+ quickTest("${mode}_${format}_one", "SELECT COUNT(1) FROM
${table}")
+ quickTest("${mode}_${format}_nullable", "SELECT COUNT(value)
FROM ${table}")
+ quickTest("${mode}_${format}_distinct", "SELECT COUNT(DISTINCT
value) FROM ${table}")
+ quickTest("${mode}_${format}_filter", "SELECT COUNT(*) FROM
${table} WHERE id > 2")
+ quickTest("${mode}_${format}_group", "SELECT value, COUNT(*)
FROM ${table} GROUP BY value", true)
+
+ if (format == "parquet" || format == "orc") {
+ // Retained assertions must see file values, not the COUNT
reader's synthetic columns.
+ def assertedCount = """SELECT COUNT(*) FROM (
+ SELECT assert_true(id > 0, 'positive id') AS checked
FROM ${table}) t"""
+ explain {
+ sql assertedCount
+ contains "pushdown agg=NONE"
+ }
+ test {
+ sql assertedCount
+ }
+ test {
+ sql """SELECT COUNT(*) FROM (
+ SELECT assert_true(id = 0, 'zero id required') AS
checked FROM ${table}) t"""
+ exception "zero id required"
+ }
+ }
+
+ // CSV has an explicit schema; empty Parquet/ORC files contain
schema metadata.
+ // A zero-byte JSON file has no schema for standalone TVF
inference.
+ if (format != "json") {
+ def emptyTable = table.replace("tvf_count_data_rows",
"tvf_count_data_empty")
+ quickTest("${mode}_${format}_empty", "SELECT COUNT(*) FROM
${emptyTable}")
+ }
+ // Two nonempty files and one empty file exercise multiple
physical ranges.
+ def multipleTable = table.replace("tvf_count_data_rows",
"tvf_count_data_*")
+ quickTest("${mode}_${format}_multiple", "SELECT COUNT(*) FROM
${multipleTable}")
+ }
+ def headerTable = """local("file_path" =
"${remotePath}/tvf_count_header.csv",
+ "backend_id" = "${backendId}", "format" = "csv_with_names",
+ "column_separator" = ",", "csv_schema" = "id:int;value:int")"""
+ // Header, empty record and final record without a newline use
normal CSV semantics.
+ quickTest("${mode}_csv_header", "SELECT COUNT(*) FROM
${headerTable}")
+ quickTest("${mode}_csv_header_nullable", "SELECT COUNT(value) FROM
${headerTable}")
+ }
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/tvf/test_s3_tvf_compression.groovy
b/regression-test/suites/external_table_p0/tvf/test_s3_tvf_compression.groovy
index 6e44cd82db9..9c32fbbcd0f 100644
---
a/regression-test/suites/external_table_p0/tvf/test_s3_tvf_compression.groovy
+++
b/regression-test/suites/external_table_p0/tvf/test_s3_tvf_compression.groovy
@@ -24,6 +24,8 @@ suite("test_s3_tvf_compression", "p0,external") {
String bucket = context.config.otherConfigs.get("s3BucketName");
sql """ set query_timeout=3600; """
+ sql """ set enable_count_push_down_for_external_table=true; """
+ sql """ set enable_push_down_no_group_agg=true; """
String compress_type = "gz"
qt_gz_1 """
@@ -53,6 +55,33 @@ suite("test_s3_tvf_compression", "p0,external") {
"compress_type" ="${compress_type}") order by cast(c1 as int),c4 limit
20;
"""
+ // Keep COUNT(*) and COUNT(1) on an unfiltered external file TVF covered by
+ // the real scan path. The expected rows are generated by the regression
+ // framework, so this also exercises compressed CSV parsing end to end.
+ qt_gz_count_star """
+ select count(*) from s3(
+ "URI" =
"https://${bucket}.${s3_endpoint}/regression/tvf/compression/test_tvf.csv.${compress_type}";,
+ "s3.access_key" = "${ak}",
+ "s3.secret_key" = "${sk}",
+ "REGION" = "${region}",
+ "FORMAT" = "csv",
+ "column_separator" = ",",
+ "use_path_style" = "false",
+ "compress_type" = "${compress_type}");
+ """
+
+ qt_gz_count_one """
+ select count(1) from s3(
+ "URI" =
"https://${bucket}.${s3_endpoint}/regression/tvf/compression/test_tvf.csv.${compress_type}";,
+ "s3.access_key" = "${ak}",
+ "s3.secret_key" = "${sk}",
+ "REGION" = "${region}",
+ "FORMAT" = "csv",
+ "column_separator" = ",",
+ "use_path_style" = "false",
+ "compress_type" = "${compress_type}");
+ """
+
compress_type = "bz2";
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
