This is an automated email from the ASF dual-hosted git repository.
FrankChen021 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 2f7bedc1e66 fix: apply the catalog's targetSegmentRows to MSQ
ingestion (#20561)
2f7bedc1e66 is described below
commit 2f7bedc1e669a7f1869b1826f9b4cebc5670069d
Author: Frank Chen <[email protected]>
AuthorDate: Sat Oct 17 23:26:32 2026 +0800
fix: apply the catalog's targetSegmentRows to MSQ ingestion (#20561)
DruidSqlValidator wrote the catalog's targetSegmentRows under
msqRowsPerSegment, a key MSQ never reads, so MSQ
ignored it and used its default. Write it under rowsPerSegment, the key MSQ
reads.
---
.../sql/calcite/planner/DruidSqlValidator.java | 4 +-
.../calcite/CalciteCatalogIngestionDmlTest.java | 52 ++++++++++++++++++++--
2 files changed, 50 insertions(+), 6 deletions(-)
diff --git
a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java
b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java
index 135b3284b4f..cf0be09c42c 100644
---
a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java
+++
b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java
@@ -93,8 +93,8 @@ public class DruidSqlValidator extends BaseDruidSqlValidator
{
private static final Pattern UNNAMED_COLUMN_PATTERN =
Pattern.compile("^EXPR\\$\\d+$", Pattern.CASE_INSENSITIVE);
- // Copied here from MSQE since that extension is not visible here.
- public static final String CTX_ROWS_PER_SEGMENT = "msqRowsPerSegment";
+ // Must match MultiStageQueryContext.CTX_ROWS_PER_SEGMENT, which is not
visible here.
+ public static final String CTX_ROWS_PER_SEGMENT = "rowsPerSegment";
private final PlannerContext plannerContext;
diff --git
a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteCatalogIngestionDmlTest.java
b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteCatalogIngestionDmlTest.java
index 4a2b99b5bb8..0c38467d316 100644
---
a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteCatalogIngestionDmlTest.java
+++
b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteCatalogIngestionDmlTest.java
@@ -40,6 +40,7 @@ import org.apache.druid.error.DruidException;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
import org.apache.druid.math.expr.ExprMacroTable;
import org.apache.druid.query.OrderBy;
import org.apache.druid.query.QueryContexts;
@@ -57,6 +58,7 @@ import
org.apache.druid.sql.calcite.external.ExternalDataSource;
import org.apache.druid.sql.calcite.external.Externals;
import org.apache.druid.sql.calcite.filtration.Filtration;
import org.apache.druid.sql.calcite.planner.CatalogResolver;
+import org.apache.druid.sql.calcite.planner.DruidSqlValidator;
import org.apache.druid.sql.calcite.run.EngineFeature;
import org.apache.druid.sql.calcite.table.DatasourceTable;
import org.apache.druid.sql.calcite.table.DruidTable;
@@ -121,7 +123,10 @@ public abstract class CalciteCatalogIngestionDmlTest
extends CalciteIngestionDml
),
new TableSpec(
DatasourceDefn.TABLE_TYPE,
-
ImmutableMap.of(DatasourceDefn.SEGMENT_GRANULARITY_PROPERTY, "PT1H"),
+ ImmutableMap.of(
+ DatasourceDefn.SEGMENT_GRANULARITY_PROPERTY,
"PT1H",
+ DatasourceDefn.TARGET_SEGMENT_ROWS_PROPERTY,
1_000_000
+ ),
ImmutableList.of(
new ColumnSpec("__time", Columns.TIME_COLUMN, null)
)
@@ -458,7 +463,7 @@ public abstract class CalciteCatalogIngestionDmlTest
extends CalciteIngestionDml
/**
* If the segment grain is given in the catalog and absent in the
PARTITIONED BY clause in the query, then use the
- * value from the catalog.
+ * value from the catalog. The catalog's target segment rows becomes the
{@code rowsPerSegment} context parameter.
*/
@Test
public void testInsertHourGrainPartitonedByFromCatalog()
@@ -475,7 +480,7 @@ public abstract class CalciteCatalogIngestionDmlTest
extends CalciteIngestionDml
.intervals(querySegmentSpec(Filtration.eternity()))
.columns("__time", "dim1", "dim2", "dim3", "cnt", "m1", "m2",
"unique_dim1")
.columnTypes(ColumnType.LONG, ColumnType.STRING,
ColumnType.STRING, ColumnType.STRING, ColumnType.LONG, ColumnType.FLOAT,
ColumnType.DOUBLE, ColumnType.ofComplex("hyperUnique"))
- .context(queryContextWithGranularity(Granularities.HOUR))
+
.context(queryContextWithGranularityAndRowsPerSegment(Granularities.HOUR,
1_000_000))
.build()
)
.verify();
@@ -501,12 +506,51 @@ public abstract class CalciteCatalogIngestionDmlTest
extends CalciteIngestionDml
.intervals(querySegmentSpec(Filtration.eternity()))
.columns("__time", "dim1", "dim2", "dim3", "cnt", "m1", "m2",
"unique_dim1")
.columnTypes(ColumnType.LONG, ColumnType.STRING,
ColumnType.STRING, ColumnType.STRING, ColumnType.LONG, ColumnType.FLOAT,
ColumnType.DOUBLE, ColumnType.ofComplex("hyperUnique"))
- .context(queryContextWithGranularity(Granularities.DAY))
+
.context(queryContextWithGranularityAndRowsPerSegment(Granularities.DAY,
1_000_000))
+ .build()
+ )
+ .verify();
+ }
+
+ /**
+ * If target segment rows is given in the catalog and {@code rowsPerSegment}
is also set in the query context, then
+ * the query context value is used.
+ */
+ @Test
+ public void testInsertRowsPerSegmentFromQueryContextOverridesCatalog()
+ {
+ final Map<String, Object> context = new HashMap<>(DEFAULT_CONTEXT);
+ context.put(DruidSqlValidator.CTX_ROWS_PER_SEGMENT, 5);
+
+ testIngestionQuery()
+ .context(context)
+ .sql(StringUtils.format(dmlPrefixPattern, "hourDs") + "\n" +
+ "SELECT * FROM foo")
+ .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+ .expectTarget("hourDs", FOO_TABLE_SIGNATURE)
+ .expectResources(dataSourceRead("hourDs"), dataSourceWrite("hourDs"),
dataSourceRead("foo"))
+ .expectQuery(
+ newScanQueryBuilder()
+ .dataSource("foo")
+ .intervals(querySegmentSpec(Filtration.eternity()))
+ .columns("__time", "dim1", "dim2", "dim3", "cnt", "m1", "m2",
"unique_dim1")
+ .columnTypes(ColumnType.LONG, ColumnType.STRING,
ColumnType.STRING, ColumnType.STRING, ColumnType.LONG, ColumnType.FLOAT,
ColumnType.DOUBLE, ColumnType.ofComplex("hyperUnique"))
+
.context(queryContextWithGranularityAndRowsPerSegment(Granularities.HOUR, 5))
.build()
)
.verify();
}
+ private Map<String, Object> queryContextWithGranularityAndRowsPerSegment(
+ final Granularity granularity,
+ final int rowsPerSegment
+ )
+ {
+ final Map<String, Object> context = new
HashMap<>(queryContextWithGranularity(granularity));
+ context.put(DruidSqlValidator.CTX_ROWS_PER_SEGMENT, rowsPerSegment);
+ return context;
+ }
+
/**
* If the segment grain is absent in the catalog and absent in the
PARTITIONED BY clause in the query, then
* validation error.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
