This is an automated email from the ASF dual-hosted git repository.
deniskuzZ pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new c82b62b11bf HIVE-30050: Sort-merge join should size a storage-handler
table from its statistics, not by listing its location (#6782)
c82b62b11bf is described below
commit c82b62b11bfc67bacbac65f193e7c9d276189566
Author: Denys Kuzmenko <[email protected]>
AuthorDate: Sat Oct 3 18:14:26 2026 +0200
HIVE-30050: Sort-merge join should size a storage-handler table from its
statistics, not by listing its location (#6782)
---
...rtitionSizeBasedBigTableSelectorForAutoSMJ.java | 6 +-
.../hive/ql/optimizer/ConvertJoinMapJoin.java | 62 ++++++------
.../SizeBasedBigTableSelectorForAutoSMJ.java | 49 ++++++++++
.../TableSizeBasedBigTableSelectorForAutoSMJ.java | 5 +-
.../TestSizeBasedBigTableSelectorForAutoSMJ.java | 108 +++++++++++++++++++++
5 files changed, 190 insertions(+), 40 deletions(-)
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/AvgPartitionSizeBasedBigTableSelectorForAutoSMJ.java
b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/AvgPartitionSizeBasedBigTableSelectorForAutoSMJ.java
index f157c6df145..9f8fe321572 100644
---
a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/AvgPartitionSizeBasedBigTableSelectorForAutoSMJ.java
+++
b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/AvgPartitionSizeBasedBigTableSelectorForAutoSMJ.java
@@ -29,7 +29,6 @@
import org.apache.hadoop.hive.ql.exec.JoinOperator;
import org.apache.hadoop.hive.ql.exec.TableScanOperator;
import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.ql.metadata.Partition;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.optimizer.ppr.PartitionPruner;
import org.apache.hadoop.hive.ql.parse.ParseContext;
@@ -82,10 +81,7 @@ public int getBigTablePosition(ParseContext parseCtx,
JoinOperator joinOp,
// For partitioned tables, get the size of all the partitions
PrunedPartitionList partsList = PartitionPruner.prune(topOp,
parseCtx, null);
numPartitions = partsList.getNotDeniedPartns().size();
- long totalSize = 0;
- for (Partition part : partsList.getNotDeniedPartns()) {
- totalSize += getSize(conf, part);
- }
+ long totalSize = getSize(conf, table,
partsList.getNotDeniedPartns());
averageSize = numPartitions == 0 ? 0 : totalSize/numPartitions;
}
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java
b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java
index 5ac61d2ec22..f6a6ff49a2e 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java
@@ -476,6 +476,24 @@ private Object
checkAndConvertSMBJoin(OptimizeTezProcContext context, JoinOperat
fallbackToReduceSideJoin(joinOp, context);
return null;
}
+ JoinDesc joinDesc = joinOp.getConf();
+ JoinCondDesc[] joinCondns = joinDesc.getConds();
+
+ Set<Integer> joinCandidates =
MapJoinProcessor.getBigTableCandidates(joinCondns);
+ if (joinCandidates.isEmpty()) {
+ // This is a full outer join. This can never be a map-join
+ // of any type. So return false.
+ return false;
+ }
+ // establish the join's structural feasibility before electing a big table:
+ // sizing the candidates can list their tables' locations on remote storage
+ if (!checkJoinSMBFeasibility(joinOp, context)) {
+ // we are just converting to a common merge join operator. The shuffle
+ // join in map-reduce case.
+ fallbackToReduceSideJoin(joinOp, context);
+ return null;
+ }
+
Class<? extends BigTableSelectorForAutoSMJ> bigTableMatcherClass = null;
try {
String selector = HiveConf.getVar(context.parseContext.getConf(),
@@ -488,14 +506,6 @@ private Object
checkAndConvertSMBJoin(OptimizeTezProcContext context, JoinOperat
BigTableSelectorForAutoSMJ bigTableMatcher =
ReflectionUtils.newInstance(bigTableMatcherClass, null);
- JoinDesc joinDesc = joinOp.getConf();
- JoinCondDesc[] joinCondns = joinDesc.getConds();
- Set<Integer> joinCandidates =
MapJoinProcessor.getBigTableCandidates(joinCondns);
- if (joinCandidates.isEmpty()) {
- // This is a full outer join. This can never be a map-join
- // of any type. So return false.
- return false;
- }
int mapJoinConversionPos =
bigTableMatcher.getBigTablePosition(context.parseContext, joinOp,
joinCandidates);
if (mapJoinConversionPos < 0) {
@@ -506,14 +516,15 @@ private Object
checkAndConvertSMBJoin(OptimizeTezProcContext context, JoinOperat
return null;
}
- if (checkConvertJoinSMBJoin(joinOp, context, mapJoinConversionPos,
tezBucketJoinProcCtx)) {
- convertJoinSMBJoin(joinOp, context, mapJoinConversionPos,
- tezBucketJoinProcCtx.getNumBuckets(), true);
- } else {
- // we are just converting to a common merge join operator. The shuffle
- // join in map-reduce case.
- fallbackToReduceSideJoin(joinOp, context);
+ ReduceSinkOperator bigTableRS =
+ (ReduceSinkOperator)
joinOp.getParentOperators().get(mapJoinConversionPos);
+ int numBuckets =
bigTableRS.getParentOperators().get(0).getOpTraits().getNumBuckets();
+
+ if (numBuckets < 0) {
+ numBuckets = bigTableRS.getConf().getNumReducers();
}
+ tezBucketJoinProcCtx.setNumBuckets(numBuckets);
+ convertJoinSMBJoin(joinOp, context, mapJoinConversionPos, numBuckets,
true);
return null;
}
@@ -803,18 +814,12 @@ private void preserveOperatorInfos(Operator<?> newOp,
Operator<?> oldOp, Optimiz
}
/*
- * This method tries to convert a join to an SMB. This is done based on
- * traits. If the sorted by columns are the same as the join columns then, we
- * can convert the join to an SMB. Otherwise retain the bucket map join as it
- * is still more efficient than a regular join.
+ * Decides from traits alone whether a join can become an SMB: the sorted by
+ * columns have to be the same as the join columns on every side. The answer
+ * does not depend on which side is elected the big table, so it is safe to
+ * consult before the election.
*/
- private boolean checkConvertJoinSMBJoin(JoinOperator joinOp,
OptimizeTezProcContext context,
- int bigTablePosition, TezBucketJoinProcCtx tezBucketJoinProcCtx) throws
SemanticException {
-
- ReduceSinkOperator bigTableRS =
- (ReduceSinkOperator) joinOp.getParentOperators().get(bigTablePosition);
- int numBuckets =
bigTableRS.getParentOperators().get(0).getOpTraits().getNumBuckets();
-
+ private boolean checkJoinSMBFeasibility(JoinOperator joinOp,
OptimizeTezProcContext context) {
int size = -1;
boolean shouldCheckExternalTables =
context.conf.getBoolVar(HiveConf.ConfVars.HIVE_DISABLE_UNSAFE_EXTERNALTABLE_OPERATIONS);
@@ -908,11 +913,6 @@ private boolean checkConvertJoinSMBJoin(JoinOperator
joinOp, OptimizeTezProcCont
}
}
- if (numBuckets < 0) {
- numBuckets = bigTableRS.getConf().getNumReducers();
- }
- tezBucketJoinProcCtx.setNumBuckets(numBuckets);
-
// With bucketing using two different versions. Version 1 for exiting
// tables and version 2 for new tables. All the inputs to the SMB must be
// from same version. This only applies to tables read directly and not
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SizeBasedBigTableSelectorForAutoSMJ.java
b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SizeBasedBigTableSelectorForAutoSMJ.java
index a0af28976ca..e1bb3e160d3 100644
---
a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SizeBasedBigTableSelectorForAutoSMJ.java
+++
b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/SizeBasedBigTableSelectorForAutoSMJ.java
@@ -23,6 +23,9 @@
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import java.util.Map;
+import org.apache.commons.lang3.math.NumberUtils;
+import org.apache.hadoop.hive.common.StatsSetupConst;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.exec.CommonJoinOperator;
import org.apache.hadoop.hive.ql.exec.Operator;
@@ -73,6 +76,11 @@ private long getSize(HiveConf conf, String size, Path path) {
}
protected long getSize(HiveConf conf, Table table) {
+ // a storage handler serves the size from its own metadata; the table's
parameters do not
+ // hold it, and listing the table's location can take minutes on object
storage
+ if (table.isNonNative()) {
+ return handlerSize(table);
+ }
Path path = table.getPath();
String size = table.getProperty("totalSize");
return getSize(conf, size, path);
@@ -84,4 +92,45 @@ protected long getSize(HiveConf conf, Partition partition) {
return getSize(conf, size, path);
}
+
+ /**
+ * Sums the sizes of the given partitions. For a storage-handler table the
partition statistics
+ * are fetched from the handler in one batch; when the handler has no
statistics for some
+ * partition the sum would fall short, so the table-level size is used
instead.
+ */
+ protected long getSize(HiveConf conf, Table table, List<Partition>
partitions) {
+ if (!table.isNonNative()) {
+ long total = 0;
+ for (Partition partition : partitions) {
+ total += getSize(conf, partition);
+ }
+ return total;
+ }
+ List<String> partNames =
partitions.stream().map(Partition::getName).toList();
+ Map<String, Map<String, String>> stats =
+ table.getStorageHandler().getAggrBasicStatsFor(table, partNames);
+ long total = 0;
+ for (String partName : partNames) {
+ Map<String, String> partStats = stats.get(partName);
+ String size = partStats != null ?
partStats.get(StatsSetupConst.TOTAL_SIZE) : null;
+ long partSize = NumberUtils.toLong(size, -1);
+ if (partSize < 0) {
+ // a partition without statistics would make the sum an underestimate,
so fall back to
+ // the whole-table size: it may overestimate the scan but never
underestimates it
+ return handlerSize(table);
+ }
+ total += partSize;
+ }
+ return total;
+ }
+
+ private static long handlerSize(Table table) {
+ if (!table.getStorageHandler().canProvideBasicStatistics()) {
+ return -1;
+ }
+ Map<String, String> stats =
table.getStorageHandler().getBasicStatistics(table);
+ return NumberUtils.toLong(
+ stats != null ? stats.get(StatsSetupConst.TOTAL_SIZE) : null,
+ -1);
+ }
}
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/TableSizeBasedBigTableSelectorForAutoSMJ.java
b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/TableSizeBasedBigTableSelectorForAutoSMJ.java
index a0c84daa1e4..76a81e70447 100644
---
a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/TableSizeBasedBigTableSelectorForAutoSMJ.java
+++
b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/TableSizeBasedBigTableSelectorForAutoSMJ.java
@@ -27,7 +27,6 @@
import org.apache.hadoop.hive.ql.exec.JoinOperator;
import org.apache.hadoop.hive.ql.exec.TableScanOperator;
import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.ql.metadata.Partition;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.optimizer.ppr.PartitionPruner;
import org.apache.hadoop.hive.ql.parse.ParseContext;
@@ -70,9 +69,7 @@ public int getBigTablePosition(ParseContext parseCtx,
JoinOperator joinOp,
else {
// For partitioned tables, get the size of all the partitions
PrunedPartitionList partsList = PartitionPruner.prune(topOp,
parseCtx, null);
- for (Partition part : partsList.getNotDeniedPartns()) {
- currentSize += getSize(conf, part);
- }
+ currentSize = getSize(conf, table, partsList.getNotDeniedPartns());
}
if (currentSize > maxSize) {
diff --git
a/ql/src/test/org/apache/hadoop/hive/ql/optimizer/TestSizeBasedBigTableSelectorForAutoSMJ.java
b/ql/src/test/org/apache/hadoop/hive/ql/optimizer/TestSizeBasedBigTableSelectorForAutoSMJ.java
new file mode 100644
index 00000000000..a3fcde0b90e
--- /dev/null
+++
b/ql/src/test/org/apache/hadoop/hive/ql/optimizer/TestSizeBasedBigTableSelectorForAutoSMJ.java
@@ -0,0 +1,108 @@
+/*
+ * 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.hadoop.hive.ql.optimizer;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.hive.common.StatsSetupConst;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.metadata.HiveStorageHandler;
+import org.apache.hadoop.hive.ql.metadata.Partition;
+import org.apache.hadoop.hive.ql.metadata.Table;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+/**
+ * The big-table choice for an automatic sort-merge join must take a handler
table's size from
+ * the handler's statistics, and must never list the table's location for one.
+ */
+public class TestSizeBasedBigTableSelectorForAutoSMJ {
+
+ private final SizeBasedBigTableSelectorForAutoSMJ selector = new
TableSizeBasedBigTableSelectorForAutoSMJ();
+ private final HiveConf conf = new HiveConf();
+
+ private Table handlerTable(Map<String, String> basicStats) {
+ HiveStorageHandler handler = Mockito.mock(HiveStorageHandler.class);
+ Mockito.when(handler.canProvideBasicStatistics()).thenReturn(true);
+
Mockito.when(handler.getBasicStatistics(Mockito.any())).thenReturn(basicStats);
+ Table table = Mockito.mock(Table.class);
+ Mockito.when(table.isNonNative()).thenReturn(true);
+ Mockito.when(table.getStorageHandler()).thenReturn(handler);
+ return table;
+ }
+
+ @Test
+ public void
handlerTableSizeComesFromItsStatisticsWithoutTouchingTheFilesystem() {
+ Table table = handlerTable(Map.of(StatsSetupConst.TOTAL_SIZE, "12345"));
+
+ Assert.assertEquals(12345, selector.getSize(conf, table));
+ // the location is only asked for on the listing fallback, which a handler
table never takes
+ Mockito.verify(table, Mockito.never()).getPath();
+ }
+
+ @Test
+ public void handlerTableOfUnknownSizeReportsUnknownRatherThanListing() {
+ Table table = handlerTable(Map.of());
+
+ Assert.assertEquals(-1, selector.getSize(conf, table));
+ Mockito.verify(table, Mockito.never()).getPath();
+ }
+
+ @Test
+ public void handlerPartitionsAreSizedTogetherAndSummed() {
+ // the table-level size covers the whole table, not one partition, so the
partitions
+ // are looked up by name in one batch and their sizes summed
+ Table table = handlerTable(Map.of(StatsSetupConst.TOTAL_SIZE, "777"));
+
Mockito.when(table.getStorageHandler().getAggrBasicStatsFor(Mockito.eq(table),
Mockito.anyList()))
+ .thenReturn(Map.of(
+ "p=a", Map.of(StatsSetupConst.TOTAL_SIZE, "100"),
+ "p=b", Map.of(StatsSetupConst.TOTAL_SIZE, "20")));
+
+ Assert.assertEquals("the partitions read, not the table once per
partition",
+ 120, selector.getSize(conf, table, List.of(partition("p=a"),
partition("p=b"))));
+ }
+
+ @Test
+ public void handlerPartitionOfUnknownSizeFallsBackToTheTable() {
+ // summing only the sized partition would underestimate the scan, so the
whole-table
+ // size is used instead: it may overestimate but never underestimates
+ Table table = handlerTable(Map.of(StatsSetupConst.TOTAL_SIZE, "777"));
+
Mockito.when(table.getStorageHandler().getAggrBasicStatsFor(Mockito.eq(table),
Mockito.anyList()))
+ .thenReturn(Map.of("p=a", Map.of(StatsSetupConst.TOTAL_SIZE, "100")));
+
+ Assert.assertEquals(777, selector.getSize(conf, table,
List.of(partition("p=a"), partition("p=b"))));
+ }
+
+ private static Partition partition(String name) {
+ Partition partition = Mockito.mock(Partition.class);
+ Mockito.when(partition.getName()).thenReturn(name);
+ return partition;
+ }
+
+ @Test
+ public void nativeTableKeepsReadingItsParameter() {
+ Table table = Mockito.mock(Table.class);
+ Mockito.when(table.isNonNative()).thenReturn(false);
+ Mockito.when(table.getProperty("totalSize")).thenReturn("4242");
+
+ Assert.assertEquals(4242, selector.getSize(conf, table));
+ }
+}
