xiangfu0 commented on code in PR #17939:
URL: https://github.com/apache/pinot/pull/17939#discussion_r4120681350
##########
pinot-common/src/main/codegen/includes/parserImpls.ftl:
##########
@@ -45,30 +45,79 @@ SqlNodeList DataFileDefList() :
}
}
+/// Parses a single VALUES row: (expr, expr, ...)
+private SqlNodeList ValuesRow() :
+{
+ SqlParserPos pos;
+ List<SqlNode> list = new ArrayList<SqlNode>();
+ SqlNode e;
+}
+{
+ <LPAREN> { pos = getPos(); }
+ e = Expression(ExprContext.ACCEPT_NON_QUERY) { list.add(e); }
+ ( <COMMA> e = Expression(ExprContext.ACCEPT_NON_QUERY) { list.add(e); } )*
+ <RPAREN>
+ {
+ return new SqlNodeList(list, pos.plus(getPos()));
+ }
+}
+
/// INSERT INTO [db_name.]table_name
-/// FROM [ FILE | ARCHIVE ] 'file_uri' [, [ FILE | ARCHIVE ] 'file_uri' ]
-SqlInsertFromFile SqlInsertFromFile() :
+/// FROM [ FILE | ARCHIVE ] 'file_uri' [, [ FILE | ARCHIVE ] 'file_uri' ]
+///
+/// INSERT INTO [db_name.]table_name [(col1, col2, ...)]
+/// VALUES (val1, val2, ...) [, (val1, val2, ...)]
+SqlNode SqlInsertStatement() :
{
SqlParserPos pos;
SqlIdentifier dbName = null;
SqlIdentifier tableName;
SqlNodeList fileList = null;
+ SqlNodeList columnList = null;
+ List<SqlNodeList> valuesList = null;
+ SqlNodeList row;
}
{
<INSERT> { pos = getPos(); }
<INTO>
[
+ LOOKAHEAD(2)
dbName = SimpleIdentifier()
<DOT>
]
tableName = SimpleIdentifier()
- [
+ (
+ // INSERT INTO t VALUES (...) without an explicit column list is not
supported.
+ // The parser rejects this shape immediately rather than building a
SqlNode that
+ // InsertIntoValues.parse would then reject — keeps error messages at
parse time.
+ LOOKAHEAD(<LPAREN>)
+ <LPAREN>
+ {
+ List<SqlNode> cols = new ArrayList<SqlNode>();
+ SqlParserPos colPos = getPos();
+ }
+ {
+ cols.add(SimpleIdentifier());
+ }
+ ( <COMMA> { cols.add(SimpleIdentifier()); } )*
+ <RPAREN>
+ <VALUES>
+ {
+ columnList = new SqlNodeList(cols, colPos.plus(getPos()));
+ valuesList = new ArrayList<SqlNodeList>();
+ }
+ row = ValuesRow() { valuesList.add(row); }
+ ( <COMMA> row = ValuesRow() { valuesList.add(row); } )*
+ {
+ return new SqlInsertIntoValues(pos, dbName, tableName, columnList,
valuesList);
+ }
+ |
fileList = DataFileDefList()
Review Comment:
Kept the new required shape — `INSERT INTO t` without FROM/VALUES now fails
at parse time instead of producing a `SqlInsertFromFile` with a null file list
that failed later at execution. Called this grammar behavior change out
explicitly in the summary comment so it's on the record.
##########
pinot-common/src/main/java/org/apache/pinot/sql/parsers/dml/InsertIntoValues.java:
##########
@@ -0,0 +1,282 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.sql.parsers.dml;
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.calcite.avatica.util.ByteString;
+import org.apache.calcite.sql.SqlLiteral;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlUnknownLiteral;
+import org.apache.calcite.util.NlsString;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.spi.config.task.AdhocTaskConfig;
+import org.apache.pinot.sql.parsers.SqlCompilationException;
+import org.apache.pinot.sql.parsers.SqlNodeAndOptions;
+import org.apache.pinot.sql.parsers.parser.SqlInsertIntoValues;
+
+
+/// DML statement for INSERT INTO ... VALUES ... syntax.
+///
+/// Parses and validates the SQL node produced by the grammar, extracting the
table name, optional column list,
+/// value rows, and options (e.g., tableType, requestId). Rejects ambiguous
hybrid table targets unless an explicit
+/// tableType option is specified.
+///
+/// This class is not thread-safe; instances are created per-request during
SQL compilation.
+public class InsertIntoValues implements DataManipulationStatement {
+
+ public static final String OPTION_TABLE_TYPE = "tableType";
+ public static final String OPTION_REQUEST_ID = "requestId";
+ public static final String TABLE_TYPE_REALTIME = "REALTIME";
+ public static final String TABLE_TYPE_OFFLINE = "OFFLINE";
+
+ private static final DataSchema INSERT_VALUES_RESPONSE_SCHEMA =
+ new DataSchema(new String[]{"statementId", "status", "message"},
+ new DataSchema.ColumnDataType[]{
+ DataSchema.ColumnDataType.STRING,
+ DataSchema.ColumnDataType.STRING,
+ DataSchema.ColumnDataType.STRING
+ });
+
+ private final String _tableName;
+ private final List<String> _columns;
+ private final List<List<Object>> _rows;
+ private final Map<String, String> _options;
+
+ /// Creates an InsertIntoValues statement.
+ ///
+ /// @param tableName resolved table name (may include database prefix)
+ /// @param columns optional column names (empty list if not specified)
+ /// @param rows list of value rows; each row is a list of literal values
+ /// @param options query options including tableType and requestId
+ public InsertIntoValues(String tableName, List<String> columns,
List<List<Object>> rows,
+ @Nullable Map<String, String> options) {
+ _tableName = tableName;
+ _columns = columns;
+ _rows = rows;
+ /// Default to an empty map so getOptionCaseInsensitive doesn't NPE for
callers that construct
+ /// InsertIntoValues without going through SqlNodeAndOptions (whose
options can also be null).
+ _options = options != null ? options : Map.of();
+ }
+
+ public String getTableName() {
+ return _tableName;
+ }
+
+ public List<String> getColumns() {
+ return _columns;
+ }
+
+ public List<List<Object>> getRows() {
+ return _rows;
+ }
+
+ public Map<String, String> getOptions() {
+ return _options;
+ }
+
+ @Nullable
+ public String getTableType() {
+ return getOptionCaseInsensitive(OPTION_TABLE_TYPE);
+ }
+
+ @Nullable
+ public String getRequestId() {
+ return getOptionCaseInsensitive(OPTION_REQUEST_ID);
+ }
+
+ @Nullable
+ private String getOptionCaseInsensitive(String key) {
+ return findOptionCaseInsensitive(_options, key);
+ }
+
+ /// Parses a {@link SqlInsertIntoValues} node into an {@link
InsertIntoValues} DML statement.
+ ///
+ /// @param sqlNodeAndOptions the parsed SQL node and options
+ /// @return the parsed InsertIntoValues statement
+ /// @throws SqlCompilationException if validation fails
+ public static InsertIntoValues parse(SqlNodeAndOptions sqlNodeAndOptions) {
+ SqlNode sqlNode = sqlNodeAndOptions.getSqlNode();
+ if (!(sqlNode instanceof SqlInsertIntoValues)) {
+ throw new SqlCompilationException("Expected SqlInsertIntoValues but got:
" + sqlNode.getClass().getSimpleName());
+ }
+ SqlInsertIntoValues insertNode = (SqlInsertIntoValues) sqlNode;
+
+ /// Resolve table name
+ String tableName = insertNode.getDbName() != null
+ ? StringUtils.joinWith(".", insertNode.getDbName(),
insertNode.getTableName())
+ : insertNode.getTableName().toString();
+
+ /// Extract column list — an explicit column list is required for INSERT
INTO VALUES.
+ /// Positional mapping (column-less INSERT) is not supported because the
SQL parser has
+ /// no access to the table schema.
+ SqlNodeList columnList = insertNode.getColumnList();
+ if (columnList == null || columnList.size() == 0) {
+ throw new SqlCompilationException(
+ "INSERT INTO ... VALUES requires an explicit column list, "
+ + "e.g. INSERT INTO myTable (col1, col2) VALUES (1, 'a'). "
+ + "Positional inserts without a column list are not supported.");
+ }
+ List<String> columns = new ArrayList<>(columnList.size());
+ for (SqlNode col : columnList) {
+ columns.add(col.toString());
+ }
+
+ /// Extract value rows
+ List<SqlNodeList> valuesList = insertNode.getValuesList();
+ if (valuesList.isEmpty()) {
+ throw new SqlCompilationException("INSERT INTO ... VALUES requires at
least one row of values");
+ }
+
+ /// Validate all rows have the same number of values
+ int expectedSize = valuesList.get(0).size();
+ if (!columns.isEmpty() && columns.size() != expectedSize) {
+ throw new SqlCompilationException(
+ "Column count (" + columns.size() + ") does not match value count ("
+ expectedSize + ")");
+ }
+
+ List<List<Object>> rows = new ArrayList<>(valuesList.size());
+ for (int i = 0; i < valuesList.size(); i++) {
+ SqlNodeList rowNode = valuesList.get(i);
+ if (rowNode.size() != expectedSize) {
+ throw new SqlCompilationException(
+ "Row " + i + " has " + rowNode.size() + " values but expected " +
expectedSize);
+ }
+ List<Object> row = new ArrayList<>(rowNode.size());
+ for (SqlNode valNode : rowNode) {
+ row.add(extractLiteralValue(valNode));
+ }
+ rows.add(row);
+ }
+
+ /// Extract options
+ Map<String, String> options = sqlNodeAndOptions.getOptions();
+
+ /// Validate tableType option if present (case-insensitive key lookup)
+ String tableType = findOptionCaseInsensitive(options, OPTION_TABLE_TYPE);
+ if (tableType != null && !TABLE_TYPE_REALTIME.equalsIgnoreCase(tableType)
+ && !TABLE_TYPE_OFFLINE.equalsIgnoreCase(tableType)) {
+ throw new SqlCompilationException(
+ "Invalid tableType '" + tableType + "'. Must be either REALTIME or
OFFLINE");
+ }
+
+ return new InsertIntoValues(tableName, columns, rows, options);
+ }
+
+ @Nullable
+ private static String findOptionCaseInsensitive(Map<String, String> options,
String key) {
+ String value = options.get(key);
+ if (value != null) {
+ return value;
+ }
+ for (Map.Entry<String, String> entry : options.entrySet()) {
+ if (entry.getKey().equalsIgnoreCase(key)) {
+ return entry.getValue();
+ }
+ }
+ return null;
+ }
+
+ /// Extracts a typed Java value from a SQL literal node.
+ ///
+ /// Returns the native Java type for each SQL literal kind:
+ /// - Numeric literals (INTEGER, DECIMAL, etc.) → {@link
java.math.BigDecimal}
Review Comment:
Fixed in adf8fb059e: `extractLiteralValue` (and the date/time-literal
rejection) now throws `SqlCompilationException` on every path, so callers
catching it for a 4xx 'bad SQL' response see these errors. Tests updated to pin
the exception type.
##########
pinot-common/src/main/java/org/apache/pinot/sql/parsers/dml/InsertIntoValues.java:
##########
@@ -0,0 +1,282 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.sql.parsers.dml;
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.calcite.avatica.util.ByteString;
+import org.apache.calcite.sql.SqlLiteral;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlUnknownLiteral;
+import org.apache.calcite.util.NlsString;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.spi.config.task.AdhocTaskConfig;
+import org.apache.pinot.sql.parsers.SqlCompilationException;
+import org.apache.pinot.sql.parsers.SqlNodeAndOptions;
+import org.apache.pinot.sql.parsers.parser.SqlInsertIntoValues;
+
+
+/// DML statement for INSERT INTO ... VALUES ... syntax.
+///
+/// Parses and validates the SQL node produced by the grammar, extracting the
table name, optional column list,
+/// value rows, and options (e.g., tableType, requestId). Rejects ambiguous
hybrid table targets unless an explicit
+/// tableType option is specified.
+///
+/// This class is not thread-safe; instances are created per-request during
SQL compilation.
+public class InsertIntoValues implements DataManipulationStatement {
+
+ public static final String OPTION_TABLE_TYPE = "tableType";
+ public static final String OPTION_REQUEST_ID = "requestId";
+ public static final String TABLE_TYPE_REALTIME = "REALTIME";
+ public static final String TABLE_TYPE_OFFLINE = "OFFLINE";
+
+ private static final DataSchema INSERT_VALUES_RESPONSE_SCHEMA =
+ new DataSchema(new String[]{"statementId", "status", "message"},
+ new DataSchema.ColumnDataType[]{
+ DataSchema.ColumnDataType.STRING,
+ DataSchema.ColumnDataType.STRING,
+ DataSchema.ColumnDataType.STRING
+ });
+
+ private final String _tableName;
+ private final List<String> _columns;
+ private final List<List<Object>> _rows;
+ private final Map<String, String> _options;
+
+ /// Creates an InsertIntoValues statement.
+ ///
+ /// @param tableName resolved table name (may include database prefix)
+ /// @param columns optional column names (empty list if not specified)
+ /// @param rows list of value rows; each row is a list of literal values
+ /// @param options query options including tableType and requestId
+ public InsertIntoValues(String tableName, List<String> columns,
List<List<Object>> rows,
+ @Nullable Map<String, String> options) {
+ _tableName = tableName;
+ _columns = columns;
+ _rows = rows;
+ /// Default to an empty map so getOptionCaseInsensitive doesn't NPE for
callers that construct
+ /// InsertIntoValues without going through SqlNodeAndOptions (whose
options can also be null).
+ _options = options != null ? options : Map.of();
+ }
+
+ public String getTableName() {
+ return _tableName;
+ }
+
+ public List<String> getColumns() {
+ return _columns;
+ }
+
+ public List<List<Object>> getRows() {
+ return _rows;
+ }
+
+ public Map<String, String> getOptions() {
+ return _options;
+ }
+
+ @Nullable
+ public String getTableType() {
+ return getOptionCaseInsensitive(OPTION_TABLE_TYPE);
+ }
+
+ @Nullable
+ public String getRequestId() {
+ return getOptionCaseInsensitive(OPTION_REQUEST_ID);
+ }
+
+ @Nullable
+ private String getOptionCaseInsensitive(String key) {
+ return findOptionCaseInsensitive(_options, key);
+ }
+
+ /// Parses a {@link SqlInsertIntoValues} node into an {@link
InsertIntoValues} DML statement.
+ ///
+ /// @param sqlNodeAndOptions the parsed SQL node and options
+ /// @return the parsed InsertIntoValues statement
+ /// @throws SqlCompilationException if validation fails
+ public static InsertIntoValues parse(SqlNodeAndOptions sqlNodeAndOptions) {
+ SqlNode sqlNode = sqlNodeAndOptions.getSqlNode();
+ if (!(sqlNode instanceof SqlInsertIntoValues)) {
+ throw new SqlCompilationException("Expected SqlInsertIntoValues but got:
" + sqlNode.getClass().getSimpleName());
+ }
+ SqlInsertIntoValues insertNode = (SqlInsertIntoValues) sqlNode;
+
+ /// Resolve table name
+ String tableName = insertNode.getDbName() != null
+ ? StringUtils.joinWith(".", insertNode.getDbName(),
insertNode.getTableName())
+ : insertNode.getTableName().toString();
+
+ /// Extract column list — an explicit column list is required for INSERT
INTO VALUES.
+ /// Positional mapping (column-less INSERT) is not supported because the
SQL parser has
+ /// no access to the table schema.
+ SqlNodeList columnList = insertNode.getColumnList();
+ if (columnList == null || columnList.size() == 0) {
+ throw new SqlCompilationException(
+ "INSERT INTO ... VALUES requires an explicit column list, "
+ + "e.g. INSERT INTO myTable (col1, col2) VALUES (1, 'a'). "
+ + "Positional inserts without a column list are not supported.");
+ }
+ List<String> columns = new ArrayList<>(columnList.size());
+ for (SqlNode col : columnList) {
+ columns.add(col.toString());
+ }
+
+ /// Extract value rows
+ List<SqlNodeList> valuesList = insertNode.getValuesList();
+ if (valuesList.isEmpty()) {
+ throw new SqlCompilationException("INSERT INTO ... VALUES requires at
least one row of values");
+ }
+
+ /// Validate all rows have the same number of values
+ int expectedSize = valuesList.get(0).size();
+ if (!columns.isEmpty() && columns.size() != expectedSize) {
+ throw new SqlCompilationException(
+ "Column count (" + columns.size() + ") does not match value count ("
+ expectedSize + ")");
+ }
+
+ List<List<Object>> rows = new ArrayList<>(valuesList.size());
+ for (int i = 0; i < valuesList.size(); i++) {
+ SqlNodeList rowNode = valuesList.get(i);
+ if (rowNode.size() != expectedSize) {
+ throw new SqlCompilationException(
+ "Row " + i + " has " + rowNode.size() + " values but expected " +
expectedSize);
+ }
+ List<Object> row = new ArrayList<>(rowNode.size());
+ for (SqlNode valNode : rowNode) {
+ row.add(extractLiteralValue(valNode));
+ }
+ rows.add(row);
+ }
+
+ /// Extract options
+ Map<String, String> options = sqlNodeAndOptions.getOptions();
+
+ /// Validate tableType option if present (case-insensitive key lookup)
+ String tableType = findOptionCaseInsensitive(options, OPTION_TABLE_TYPE);
+ if (tableType != null && !TABLE_TYPE_REALTIME.equalsIgnoreCase(tableType)
+ && !TABLE_TYPE_OFFLINE.equalsIgnoreCase(tableType)) {
+ throw new SqlCompilationException(
+ "Invalid tableType '" + tableType + "'. Must be either REALTIME or
OFFLINE");
+ }
+
+ return new InsertIntoValues(tableName, columns, rows, options);
+ }
+
+ @Nullable
+ private static String findOptionCaseInsensitive(Map<String, String> options,
String key) {
+ String value = options.get(key);
+ if (value != null) {
+ return value;
+ }
+ for (Map.Entry<String, String> entry : options.entrySet()) {
+ if (entry.getKey().equalsIgnoreCase(key)) {
+ return entry.getValue();
+ }
+ }
+ return null;
+ }
+
+ /// Extracts a typed Java value from a SQL literal node.
+ ///
+ /// Returns the native Java type for each SQL literal kind:
+ /// - Numeric literals (INTEGER, DECIMAL, etc.) → {@link
java.math.BigDecimal}
+ /// - Boolean literals → {@link Boolean}
+ /// - String/char literals → {@link String} (unwrapped from Calcite's {@link
NlsString})
+ /// - NULL → `null`
+ ///
+ /// Date, time, and timestamp literals (e.g. `DATE '2024-01-01'`,
+ /// `TIMESTAMP '2024-01-01 00:00:00'`) are explicitly rejected. In Pinot's
Calcite grammar
+ /// these are parsed as {@link SqlUnknownLiteral} nodes; the schema-aware
coercion path that
+ /// converts them to epoch millis does not exist yet on the INSERT INTO
VALUES path.
+ /// Use epoch-millis integers (e.g. `1704067200000`) or ISO-8601 strings
instead.
+ private static Object extractLiteralValue(SqlNode node) {
+ if (node instanceof SqlLiteral) {
+ SqlLiteral literal = (SqlLiteral) node;
+ Object value = literal.getValue();
+ if (value == null) {
+ return null;
+ }
+ /// Reject date/time literal types — no downstream coercion to Pinot
column types exists yet.
+ /// Pinot's SQL grammar creates SqlUnknownLiteral (typeName=UNKNOWN) for
DATE/TIME/TIMESTAMP
+ /// prefix syntax (e.g. DATE '2024-01-01'), not typed date literals.
Detect these by checking
+ /// the literal subtype and its tag.
+ if (literal instanceof SqlUnknownLiteral) {
+ String tag = ((SqlUnknownLiteral) literal).tag;
+ if (tag != null) {
+ String upperTag = tag.toUpperCase(Locale.ROOT);
+ if ("DATE".equals(upperTag) || "TIME".equals(upperTag) ||
"TIMESTAMP".equals(upperTag)) {
+ throw new IllegalArgumentException(
+ "Date/time literals are not supported in INSERT INTO ...
VALUES (no schema-aware coercion). "
+ + "Use epoch-millis integers or ISO-8601 strings instead.
Got: " + node);
+ }
+ }
+ }
+ /// Unwrap Calcite's NlsString to a plain Java String for char/varchar
literals.
+ if (value instanceof NlsString) {
+ return ((NlsString) value).getValue();
+ }
+ /// Unwrap Calcite's ByteString to byte[] for X'AB12' BYTES literals so
Pinot BYTES columns
+ /// can be inserted via INSERT INTO VALUES.
+ if (value instanceof ByteString) {
+ return ((ByteString) value).getBytes();
+ }
+ /// Whitelist supported value types so INTERVAL,
TIMESTAMP_WITH_LOCAL_TIME_ZONE, and other
+ /// exotic Calcite literal subtypes don't silently fall through with
their raw toString.
+ if (value instanceof BigDecimal || value instanceof Boolean || value
instanceof String
+ || value instanceof byte[]) {
+ return value;
+ }
+ throw new IllegalArgumentException(
+ "Unsupported literal type in INSERT INTO ... VALUES: " +
value.getClass().getSimpleName()
+ + " (value: " + value + "). Supported types: numeric, boolean,
character/varchar, bytes.");
+ }
+ /// Non-literal expressions (e.g. 1+1, CURRENT_TIMESTAMP) are not
supported in VALUES
+ throw new IllegalArgumentException(
+ "Only literal values are supported in INSERT INTO ... VALUES. "
Review Comment:
Fixed in adf8fb059e: MINUS_PREFIX/PLUS_PREFIX around numeric literals are
unwrapped (mirroring CalciteSqlParser's handling), so `VALUES (-1)`, `(-2.5)`,
`(+3)` parse. Unary minus on a non-numeric literal gets a clear
`SqlCompilationException`. Added `testNegativeAndSignedNumericLiterals` and
`testUnaryMinusOnNonNumericLiteralIsRejected`, plus a negative-values case in
the cluster IT.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java:
##########
@@ -702,6 +756,31 @@ private void setUpPinotController() {
// Setting up periodic tasks
List<PeriodicTask> controllerPeriodicTasks =
setupControllerPeriodicTasks();
+ // Register FILE executor now that _taskManager is available (created
during periodic task setup).
+ // Pass _helixTaskResourceManager so the executor can poll Minion task
state to auto-complete inserts.
+ // Then start() the coordinator — only now is isStarted() flipped to true
and the REST surface
+ // unlocked. Until this line a ROW or FILE INSERT request would have hit
checkEnabled()'s 503
+ // because isStarted()=false, avoiding the "ROW works but FILE returns
NO_EXECUTOR" window.
+ if (_config.isInsertEnabled()) {
+ if (_taskManager == null) {
+ // No Minion task manager configured — register only the ROW executor
and start the
+ // coordinator. ROW inserts (INSERT INTO VALUES) are designed for
interactive / quickstart
+ // workloads and don't require Minion. FILE inserts will be rejected
with NO_EXECUTOR by
+ // the coordinator's existing executor lookup; operators who need FILE
support must enable
+ // Minion task management. Log at ERROR — partial-feature state is a
misconfiguration that
+ // operators should notice immediately rather than discover when an
INSERT FROM FILE fails.
+ LOGGER.error("controller.insert.enabled=true but PinotTaskManager is
not configured. "
+ + "ROW inserts will work but FILE inserts will be rejected with
NO_EXECUTOR. "
+ + "To enable FILE inserts, configure Minion task management;
otherwise set "
+ + "controller.insert.enabled=false to make the partial-feature
state explicit.");
+ } else {
+ FileInsertExecutor fileInsertExecutor =
+ new FileInsertExecutor(_helixResourceManager, _taskManager,
_helixTaskResourceManager);
+ _insertStatementCoordinator.registerExecutor(InsertType.FILE.name(),
fileInsertExecutor);
+ }
+ _insertStatementCoordinator.start();
Review Comment:
Fixed in adf8fb059e: the coordinator now takes the `LeadControllerManager`
and the cleanup sweep skips tables this controller does not lead
(`isLeaderForTable`), so N flag-enabled controllers no longer run duplicate
sweeps, race on CAS transitions, or each poll Minion state. Tables stay in
`_tablesWithStatements` so a leadership change picks them up on the next sweep.
Routing `/insert/execute` to the table's lead controller is a fair follow-up —
noted in the summary.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/ingest/ControllerRowInsertExecutor.java:
##########
@@ -0,0 +1,566 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.helix.core.ingest;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.math.BigDecimal;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.URIUtils;
+import org.apache.pinot.controller.api.resources.ControllerFilePathProvider;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.PinotResourceManagerResponse;
+import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.spi.SegmentMetadata;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
+import org.apache.pinot.segment.spi.partition.PartitionFunction;
+import org.apache.pinot.segment.spi.partition.PartitionFunctionFactory;
+import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.IndexingConfig;
+import org.apache.pinot.spi.config.table.SegmentPartitionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.config.table.UpsertConfig;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.ingest.InsertErrorCode;
+import org.apache.pinot.spi.ingest.InsertExecutor;
+import org.apache.pinot.spi.ingest.InsertRequest;
+import org.apache.pinot.spi.ingest.InsertResult;
+import org.apache.pinot.spi.ingest.InsertStatementState;
+import org.apache.pinot.spi.utils.BytesUtils;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Controller-side INSERT INTO ... VALUES executor that builds a Pinot
segment from in-memory
+/// rows and uploads it directly to the segment store.
+///
+/// For OFFLINE tables the segment is generated and uploaded immediately. For
REALTIME tables
+/// the same approach is used (Pinot supports offline-style segment upload for
backfill).
+///
+/// This executor is designed for interactive / quickstart use cases where low
row counts are
+/// inserted via SQL. It is not intended for high-throughput production
ingestion.
+///
+/// Instances are thread-safe; each {@link #execute} call uses its own
temporary directory.
+///
+/// ## Known v1 limitation: single-phase visibility
+///
+/// {@link #execute} uploads segments to the deep store and registers them in
Helix IdealState
+/// within a single call, returning {@link InsertStatementState#VISIBLE}
before the coordinator has
+/// persisted that state to ZK. If the subsequent manifest persist fails after
{@code
+/// persistWithCasRetry}'s three CAS attempts, the data is queryable but the
manifest is stuck in
+/// `ACCEPTED`. The cleanup sweep deliberately does NOT abort such
stuck-ACCEPTED ROW
+/// manifests — flipping them to ABORTED would falsely report failure for live
data. Instead, the
+/// sweep GCs them after a long `ACCEPTED_ROW_RETENTION_MS` window (default 7
days), giving
+/// operators time to investigate. Until GC, `/insert/status` will report
state=ACCEPTED.
+///
+/// A proper two-phase protocol (stage-without-register, then finalize on
coordinator signal)
+/// is a follow-up. Documented so operators can reason about the tradeoff:
duplicate data is
+/// prevented (we never release the requestId on persist failure), but metric
counters and UI
+/// status may lag the actual visible state in rare ZK-down-for-minutes
scenarios.
+public class ControllerRowInsertExecutor implements InsertExecutor {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ControllerRowInsertExecutor.class);
+
+ private static final String SEGMENT_NAME_PREFIX = "insert_";
+ private static final String WORKING_DIR_PREFIX = "row_insert_";
+ private static final String OUTPUT_SEGMENT_DIR = "output_segment";
+ private static final String SEGMENT_TAR_DIR = "segment_tar";
+
+ private final PinotHelixResourceManager _resourceManager;
+ private final boolean _allowDestructiveRollback;
+
+ /// Creates a new executor with destructive rollback DISABLED. Partial-batch
upload failures
+ /// leave already-registered segments in IdealState; operators reconcile
manually.
+ public ControllerRowInsertExecutor(PinotHelixResourceManager
resourceManager) {
+ this(resourceManager, false);
+ }
+
+ /// Creates a new executor.
+ ///
+ /// @param resourceManager the Helix resource manager for table
config lookup and segment upload
+ /// @param allowDestructiveRollback if `true`, partial-batch upload failure
deletes
+ /// already-registered segments via `_resourceManager.deleteSegment`.
This can yank
+ /// segments out from under live queries and is OFF by default. Set via
+ /// `controller.insert.row.allow.destructive.rollback` on
interactive/quickstart-tier
+ /// controllers only.
+ public ControllerRowInsertExecutor(PinotHelixResourceManager
resourceManager, boolean allowDestructiveRollback) {
+ _resourceManager = resourceManager;
+ _allowDestructiveRollback = allowDestructiveRollback;
+ }
+
+ @Override
+ public InsertResult execute(InsertRequest request) {
Review Comment:
Fixed in adf8fb059e: three bounds, all configurable —
`controller.insert.row.max.rows.per.statement` (default 10k) and
`controller.insert.row.max.bytes.per.statement` (default 10MB) are enforced in
the coordinator **before a manifest is created** (REJECTED with
ROW_LIMIT_EXCEEDED / PAYLOAD_TOO_LARGE), and
`controller.insert.row.max.concurrent.statements` (default 4) is a fail-fast
semaphore around the segment build (TOO_MANY_CONCURRENT_INSERTS, retryable).
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/ingest/ControllerRowInsertExecutor.java:
##########
@@ -0,0 +1,566 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.helix.core.ingest;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.File;
+import java.math.BigDecimal;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import javax.annotation.Nullable;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.utils.TarCompressionUtils;
+import org.apache.pinot.common.utils.URIUtils;
+import org.apache.pinot.controller.api.resources.ControllerFilePathProvider;
+import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
+import org.apache.pinot.controller.helix.core.PinotResourceManagerResponse;
+import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.spi.SegmentMetadata;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
+import org.apache.pinot.segment.spi.partition.PartitionFunction;
+import org.apache.pinot.segment.spi.partition.PartitionFunctionFactory;
+import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.IndexingConfig;
+import org.apache.pinot.spi.config.table.SegmentPartitionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.config.table.UpsertConfig;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.filesystem.PinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.apache.pinot.spi.ingest.InsertErrorCode;
+import org.apache.pinot.spi.ingest.InsertExecutor;
+import org.apache.pinot.spi.ingest.InsertRequest;
+import org.apache.pinot.spi.ingest.InsertResult;
+import org.apache.pinot.spi.ingest.InsertStatementState;
+import org.apache.pinot.spi.utils.BytesUtils;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Controller-side INSERT INTO ... VALUES executor that builds a Pinot
segment from in-memory
+/// rows and uploads it directly to the segment store.
+///
+/// For OFFLINE tables the segment is generated and uploaded immediately. For
REALTIME tables
+/// the same approach is used (Pinot supports offline-style segment upload for
backfill).
+///
+/// This executor is designed for interactive / quickstart use cases where low
row counts are
+/// inserted via SQL. It is not intended for high-throughput production
ingestion.
+///
+/// Instances are thread-safe; each {@link #execute} call uses its own
temporary directory.
+///
+/// ## Known v1 limitation: single-phase visibility
+///
+/// {@link #execute} uploads segments to the deep store and registers them in
Helix IdealState
+/// within a single call, returning {@link InsertStatementState#VISIBLE}
before the coordinator has
+/// persisted that state to ZK. If the subsequent manifest persist fails after
{@code
+/// persistWithCasRetry}'s three CAS attempts, the data is queryable but the
manifest is stuck in
+/// `ACCEPTED`. The cleanup sweep deliberately does NOT abort such
stuck-ACCEPTED ROW
+/// manifests — flipping them to ABORTED would falsely report failure for live
data. Instead, the
+/// sweep GCs them after a long `ACCEPTED_ROW_RETENTION_MS` window (default 7
days), giving
+/// operators time to investigate. Until GC, `/insert/status` will report
state=ACCEPTED.
+///
+/// A proper two-phase protocol (stage-without-register, then finalize on
coordinator signal)
+/// is a follow-up. Documented so operators can reason about the tradeoff:
duplicate data is
+/// prevented (we never release the requestId on persist failure), but metric
counters and UI
+/// status may lag the actual visible state in rare ZK-down-for-minutes
scenarios.
+public class ControllerRowInsertExecutor implements InsertExecutor {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ControllerRowInsertExecutor.class);
+
+ private static final String SEGMENT_NAME_PREFIX = "insert_";
+ private static final String WORKING_DIR_PREFIX = "row_insert_";
+ private static final String OUTPUT_SEGMENT_DIR = "output_segment";
+ private static final String SEGMENT_TAR_DIR = "segment_tar";
+
+ private final PinotHelixResourceManager _resourceManager;
+ private final boolean _allowDestructiveRollback;
+
+ /// Creates a new executor with destructive rollback DISABLED. Partial-batch
upload failures
+ /// leave already-registered segments in IdealState; operators reconcile
manually.
+ public ControllerRowInsertExecutor(PinotHelixResourceManager
resourceManager) {
+ this(resourceManager, false);
+ }
+
+ /// Creates a new executor.
+ ///
+ /// @param resourceManager the Helix resource manager for table
config lookup and segment upload
+ /// @param allowDestructiveRollback if `true`, partial-batch upload failure
deletes
+ /// already-registered segments via `_resourceManager.deleteSegment`.
This can yank
+ /// segments out from under live queries and is OFF by default. Set via
+ /// `controller.insert.row.allow.destructive.rollback` on
interactive/quickstart-tier
+ /// controllers only.
+ public ControllerRowInsertExecutor(PinotHelixResourceManager
resourceManager, boolean allowDestructiveRollback) {
+ _resourceManager = resourceManager;
+ _allowDestructiveRollback = allowDestructiveRollback;
+ }
+
+ @Override
+ public InsertResult execute(InsertRequest request) {
+ String statementId = request.getStatementId();
+ String tableName = request.getTableName();
+ TableType tableType = request.getTableType();
+ List<GenericRow> rows = request.getRows();
+
+ LOGGER.info("Executing row insert statement {} for table {} with {} rows",
statementId, tableName,
+ rows == null ? 0 : rows.size());
+
+ /// 1. Validate rows. The coordinator's submit path pre-rejects empty rows
for ROW with
+ /// EMPTY_ROWS and state=REJECTED (no manifest persisted) — that's the
path real callers hit.
+ /// This branch is defense-in-depth for direct executor.execute callers
(tests, future
+ /// programmatic SPI consumers) that bypass the coordinator; it returns
state=ABORTED via
+ /// buildErrorResult because by convention REJECTED is reserved for
coordinator-tier pre-
+ /// acceptance rejections (no-manifest semantics). In v1 an executor only
reports terminal
+ /// states for which a manifest exists, i.e., ABORTED/VISIBLE (future
async paths may emit
+ /// ACCEPTED, but the coordinator owns the lifecycle past that point).
+ if (rows == null || rows.isEmpty()) {
+ return buildErrorResult(statementId, "No rows provided in the insert
request.", InsertErrorCode.EMPTY_ROWS);
+ }
+
+ /// 2. Resolve the fully-qualified table name
+ String tableNameWithType = resolveTableNameWithType(tableName, tableType);
+
+ /// 3. Look up table config and schema
+ TableConfig tableConfig =
_resourceManager.getTableConfig(tableNameWithType);
+ if (tableConfig == null) {
+ return buildErrorResult(statementId, "Table not found: " +
tableNameWithType, InsertErrorCode.TABLE_NOT_FOUND);
+ }
+
+ /// 3a. Validate table mode safety (same rules as FileInsertExecutor)
+ String safetyError = validateTableModeSafety(tableConfig);
+ if (safetyError != null) {
+ return buildErrorResult(statementId, safetyError,
InsertErrorCode.TABLE_MODE_REJECTED);
+ }
+
+ Schema schema = _resourceManager.getTableSchema(tableNameWithType);
+ if (schema == null) {
+ return buildErrorResult(statementId, "Schema not found for table: " +
tableNameWithType,
+ InsertErrorCode.SCHEMA_NOT_FOUND);
+ }
+
+ /// 3b. Reject null primary-key values for full upsert tables. PK columns
are tracked on the
+ /// schema, NOT on the partition config — a previous version of this code
conflated the two.
+ String pkValidationError = validatePrimaryKeyValues(rows, tableConfig,
schema);
+ if (pkValidationError != null) {
+ return buildErrorResult(statementId, pkValidationError,
InsertErrorCode.PRIMARY_KEY_REJECTED);
+ }
+
+ /// 4. Split rows by partition to produce correctly partitioned segments
+ boolean hasPartitionConfig = tableConfig.getIndexingConfig() != null
+ && tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null
+ &&
!tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap().isEmpty();
+ Map<Integer, List<GenericRow>> partitionedRows;
+ try {
+ partitionedRows = partitionRows(rows, tableConfig, schema);
+ } catch (IllegalArgumentException e) {
+ return buildErrorResult(statementId, e.getMessage(),
InsertErrorCode.PARTITION_VALUE_REJECTED);
+ }
+
+ /// 5. Build all segments first (staging phase), then upload all
atomically.
+ /// If any build fails, no segments have been uploaded yet so nothing to
roll back.
+ /// If upload fails partway through, we roll back already-uploaded
segments.
+ File workingDir = null;
+ try {
+ workingDir = createWorkingDir(tableNameWithType);
+ FileUtils.forceMkdir(workingDir);
+
+ /// Phase 1: Build all segments locally without uploading
+ List<StagedSegment> stagedSegments = new ArrayList<>();
+
+ for (Map.Entry<Integer, List<GenericRow>> entry :
partitionedRows.entrySet()) {
+ int partitionId = entry.getKey();
+ List<GenericRow> partitionRows = entry.getValue();
+
+ /// Always include the _p<partitionId> suffix when the table has a
partition config, even if
+ /// all rows in this batch happen to land on a single partition.
Otherwise downstream tooling
+ /// that parses the suffix (partition-aware purge tasks, segment
lineage by partition) sees
+ /// an unsuffixed name and falsely concludes the segment is
unpartitioned.
+ String segmentName = hasPartitionConfig
+ ? SEGMENT_NAME_PREFIX + statementId + "_p" + partitionId
+ : SEGMENT_NAME_PREFIX + statementId;
+
+ File partitionDir = new File(workingDir, "partition_" + partitionId);
+ File outputDir = new File(partitionDir, OUTPUT_SEGMENT_DIR);
+ File segmentTarDir = new File(partitionDir, SEGMENT_TAR_DIR);
+ FileUtils.forceMkdir(outputDir);
+ FileUtils.forceMkdir(segmentTarDir);
+
+ SegmentGeneratorConfig segmentGeneratorConfig = new
SegmentGeneratorConfig(tableConfig, schema);
+ segmentGeneratorConfig.setOutDir(outputDir.getAbsolutePath());
+ segmentGeneratorConfig.setSegmentName(segmentName);
+ segmentGeneratorConfig.setTableName(tableNameWithType);
+
+ GenericRowRecordReader recordReader = new
GenericRowRecordReader(partitionRows);
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(segmentGeneratorConfig, recordReader);
+ driver.build();
+
+ File segmentOutputDir = driver.getOutputDirectory();
+ File segmentTarFile = new File(segmentTarDir, segmentName + ".tar.gz");
+ TarCompressionUtils.createCompressedTarFile(segmentOutputDir,
segmentTarFile);
+
+ stagedSegments.add(new StagedSegment(segmentName, segmentOutputDir,
segmentTarFile, partitionRows.size(),
+ partitionId));
+
+ LOGGER.info("Staged segment {} ({} rows, partition {})", segmentName,
partitionRows.size(), partitionId);
+ }
+
+ /// Phase 2: Upload all staged segments. Track what was fully uploaded
(deep-store copy AND
+ /// Helix register both succeeded) so rollback can call deleteSegment on
registered segments.
+ /// The failing segment is tracked separately (`pendingSegmentName`) so
the manifest doesn't
+ /// claim phantom uploads that never landed in IdealState.
+ List<String> uploadedSegmentNames = new ArrayList<>();
+ String pendingSegmentName = null;
+ try {
+ for (StagedSegment staged : stagedSegments) {
+ pendingSegmentName = staged._segmentName;
+ uploadSegment(tableNameWithType, staged._segmentDir,
staged._tarFile, staged._segmentName);
+ /// Add to the success list ONLY after uploadSegment returns: at
that point both the
+ /// deep-store copy and the Helix register succeeded. A
pre-add-then-throw would put a
+ /// phantom name in the manifest's segmentNames if rollback failed.
+ uploadedSegmentNames.add(staged._segmentName);
+ pendingSegmentName = null;
+ LOGGER.info("Uploaded segment {} ({} rows, partition {})",
+ staged._segmentName, staged._rowCount, staged._partitionId);
+ }
+ } catch (Exception uploadEx) {
+ if (_allowDestructiveRollback) {
+ LOGGER.error("Failed to upload segment during row insert for
statement {} on table {}. "
+ + "Rolling back {} already-uploaded segments (destructive
rollback enabled). Pending "
+ + "segment {} may have left orphan tar in deep store (operator
cleanup may be required).",
+ statementId, tableNameWithType, uploadedSegmentNames.size(),
pendingSegmentName, uploadEx);
+ rollbackUploadedSegments(tableNameWithType, uploadedSegmentNames);
+ return buildErrorResult(statementId,
+ "Failed to upload segment (rolled back " +
uploadedSegmentNames.size() + " segments): "
+ + uploadEx.getMessage(),
+ InsertErrorCode.SEGMENT_UPLOAD_FAILED);
+ }
+ /// Default: do NOT call deleteSegment on already-registered segments
— that would yank
+ /// segments out from under live queries on the table. Surface the
partial set in the
+ /// result so operators can reconcile manually. The pending segment
(if any) may have
+ /// left an orphan tar in deep store and an unregistered segment in
IdealState.
+ LOGGER.error("Failed to upload segment during row insert for statement
{} on table {}. "
+ + "Leaving {} already-uploaded segments registered (destructive
rollback disabled). "
+ + "Operator must reconcile. Pending segment {} may have left
orphan tar in deep store.",
+ statementId, tableNameWithType, uploadedSegmentNames.size(),
pendingSegmentName, uploadEx);
+ return new InsertResult.Builder()
+ .setStatementId(statementId)
+ .setState(InsertStatementState.ABORTED)
+ .setMessage("Failed to upload segment after registering " +
uploadedSegmentNames.size()
+ + " segments; left in IdealState for operator reconciliation:
" + uploadEx.getMessage())
+ .setErrorCode(InsertErrorCode.SEGMENT_UPLOAD_FAILED_PARTIAL)
+ .setSegmentNames(uploadedSegmentNames)
+ .build();
+ }
+
+ return new InsertResult.Builder()
+ .setStatementId(statementId)
+ .setState(InsertStatementState.VISIBLE)
+ .setMessage("Successfully inserted " + rows.size() + " rows as " +
uploadedSegmentNames.size()
+ + " segment(s)")
+ .setSegmentNames(uploadedSegmentNames)
+ .build();
+ } catch (Exception e) {
+ LOGGER.error("Failed to execute row insert for statement {} on table
{}", statementId, tableNameWithType, e);
+ return buildErrorResult(statementId, "Failed to build segment: " +
e.getMessage(),
+ InsertErrorCode.SEGMENT_BUILD_FAILED);
+ } finally {
+ FileUtils.deleteQuietly(workingDir);
+ }
+ }
+
+ @Override
+ public void abort(String statementId) {
+ /// Row inserts are synchronous and complete within execute(), so abort is
a no-op. A trace log
+ /// is included so a future maintainer who wires asynchronous row inserts
can spot stray abort
+ /// calls reaching here without needing to add the log themselves. If
trace logging shows up,
+ /// re-evaluate the no-op contract before adding rollback logic — the
manifest state machine
+ /// already handles abort at the coordinator layer.
+ LOGGER.trace("ControllerRowInsertExecutor.abort({}): no-op for synchronous
ROW path", statementId);
+ }
+
+ /// Partitions rows by the table's segment partition config. If no partition
config exists,
+ /// all rows are placed in partition 0 (single segment).
+ ///
+ /// Partition values are coerced to the column's declared data type before
hashing so that
+ /// INSERT-time partitioning matches query-time partitioning (which coerces
predicate values to
+ /// the declared type). Without coercion, logically-equivalent values (e.g.,
`BigDecimal(3.0)`
+ /// from a SQL literal versus `Long(3)` from a query predicate) would hash
to different
+ /// partitions, causing query predicate pruning to skip segments that
actually contain matching rows.
+ ///
+ /// For full upsert tables, null primary keys are rejected up-front since
they violate the
+ /// upsert invariant.
+ private Map<Integer, List<GenericRow>> partitionRows(List<GenericRow> rows,
TableConfig tableConfig,
+ Schema schema) {
+ IndexingConfig indexingConfig = tableConfig.getIndexingConfig();
+ SegmentPartitionConfig partitionConfig = indexingConfig != null
+ ? indexingConfig.getSegmentPartitionConfig() : null;
+
+ if (partitionConfig == null || partitionConfig.getColumnPartitionMap() ==
null
+ || partitionConfig.getColumnPartitionMap().isEmpty()) {
+ /// No partition config — single segment
+ return Collections.singletonMap(0, rows);
+ }
+
+ /// Multi-column configs are rejected by validateTableModeSafety before
this point.
+ Map<String, ColumnPartitionConfig> columnPartitionMap =
partitionConfig.getColumnPartitionMap();
+ Map.Entry<String, ColumnPartitionConfig> entry =
columnPartitionMap.entrySet().iterator().next();
+ String partitionColumn = entry.getKey();
+ ColumnPartitionConfig colConfig = entry.getValue();
+ int numPartitions = colConfig.getNumPartitions();
+ PartitionFunction partitionFunction =
PartitionFunctionFactory.getPartitionFunction(
+ colConfig.getFunctionName(), numPartitions,
colConfig.getFunctionConfig());
+
+ FieldSpec fieldSpec = schema.getFieldSpecFor(partitionColumn);
+ FieldSpec.DataType storedType = fieldSpec != null ?
fieldSpec.getDataType().getStoredType() : null;
+
+ Map<Integer, List<GenericRow>> partitioned = new HashMap<>();
+ for (GenericRow row : rows) {
+ Object value = row.getValue(partitionColumn);
+ if (value == null) {
+ /// Match server-side stream behavior: hash "" through the partition
function rather than
+ /// defaulting to partition 0. Primary-key null rejection is handled
separately in
+ /// validatePrimaryKeyValues, before this method is called, so we
don't repeat it here.
+ int partition = partitionFunction.getPartition("");
+ partitioned.computeIfAbsent(partition, k -> new
ArrayList<>()).add(row);
+ continue;
+ }
+ String partitionValue = normalizeForPartition(value, storedType);
+ int partition = partitionFunction.getPartition(partitionValue);
+ partitioned.computeIfAbsent(partition, k -> new ArrayList<>()).add(row);
+ }
+ return partitioned;
+ }
+
+ /// Validates that no row has a null value in any primary-key column when
the table is in full
+ /// upsert mode. Primary-key columns are configured on the schema, not on
the partition config.
+ /// Returns an error message if any row violates the invariant, or `null` if
all rows are OK
+ /// (or the table is not a full-upsert table).
+ @Nullable
+ @VisibleForTesting
+ static String validatePrimaryKeyValues(List<GenericRow> rows, TableConfig
tableConfig, Schema schema) {
+ UpsertConfig upsertConfig = tableConfig.getUpsertConfig();
+ if (upsertConfig == null || upsertConfig.getMode() !=
UpsertConfig.Mode.FULL) {
+ return null;
+ }
+ List<String> pkColumns = schema.getPrimaryKeyColumns();
+ if (pkColumns == null || pkColumns.isEmpty()) {
+ /// Schema validation should have rejected this earlier, but fail loudly
rather than silently
+ /// accepting a full-upsert table with no PK columns.
+ return "Full upsert table is missing primary-key columns in schema";
+ }
+ for (int rowIdx = 0; rowIdx < rows.size(); rowIdx++) {
+ GenericRow row = rows.get(rowIdx);
+ for (String pkColumn : pkColumns) {
+ if (row.getValue(pkColumn) == null) {
+ return "Null value in primary-key column '" + pkColumn + "' is not
allowed for full upsert tables "
+ + "(row index " + rowIdx + ")";
+ }
+ }
+ }
+ return null;
+ }
+
+ /// Coerces a value to the column's declared stored type then stringifies
it, so that
+ /// logically-equivalent values hash to the same partition regardless of the
source literal's
+ /// Java type. Falls back to raw stringification if the column's type is
unknown or the coercion
+ /// fails (bad data is caught later during segment build).
+ @VisibleForTesting
+ static String normalizeForPartition(Object value, @Nullable
FieldSpec.DataType storedType) {
+ if (storedType == null) {
+ /// The schema lookup at the call site should have already resolved the
column's type. A null
+ /// storedType here means the partition column is missing from the
schema — coercing through
+ /// String.valueOf would silently mis-route the row vs the query path's
typed partition lookup,
+ /// ending in segment-pruning at query time. Fail loudly so the operator
sees the schema bug.
+ throw new IllegalArgumentException(
+ "Cannot normalize partition value: column has no declared type in
schema (value=" + value + ")");
+ }
+ /// Canonicalize numeric inputs before String conversion. The SQL parser
produces BigDecimal for
+ /// every numeric literal, so a literal like `3.0` arrives as
BigDecimal("3.0"). Without
+ /// stripTrailingZeros, String.valueOf yields "3.0" which LONG.convert
rejects with
+ /// NumberFormatException — even though `INSERT INTO t (id) VALUES (3.0)`
is intended as the
+ /// value 3 in a LONG-partitioned table. Strip trailing zeros so
integral-typed coercion sees
+ /// the canonical integer form. Genuinely-fractional values (3.5) still
fail at convert time.
+ String stringForm;
+ if (value instanceof BigDecimal) {
+ stringForm = ((BigDecimal) value).stripTrailingZeros().toPlainString();
+ } else {
+ stringForm = String.valueOf(value);
+ }
+ Object coerced;
+ try {
+ coerced = storedType.convert(stringForm);
+ } catch (Exception e) {
+ /// Don't silently degrade to raw String.valueOf — that risks hashing a
malformed value into a
+ /// partition where logically-equivalent well-typed values would land
elsewhere, and the row
+ /// gets mis-routed before segment-build catches the type error. Fail
loudly so the caller
+ /// surfaces PARTITION_VALUE_REJECTED.
+ throw new IllegalArgumentException(
+ "Could not coerce value=" + value + " to declared partition-column
type " + storedType
+ + ": " + e.getMessage(), e);
+ }
+ if (coerced instanceof byte[]) {
+ return BytesUtils.toHexString((byte[]) coerced);
+ }
+ return String.valueOf(coerced);
+ }
+
+ /// Creates the temporary working directory for segment build.
Package-private for test overriding.
+ File createWorkingDir(String tableNameWithType) {
+ ControllerFilePathProvider pathProvider =
ControllerFilePathProvider.getInstance();
+ String uniqueSuffix = UUID.randomUUID().toString().replace("-",
"").substring(0, 12);
+ return new File(pathProvider.getFileUploadTempDir(),
+ WORKING_DIR_PREFIX + tableNameWithType + "_" + uniqueSuffix);
+ }
+
+ /// Uploads the segment tar file to the controller deep store and registers
it in ZooKeeper
+ /// via the resource manager. Package-private for test overriding.
+ void uploadSegment(String tableNameWithType, File segmentDir, File
segmentTarFile,
Review Comment:
Fixed in adf8fb059e: the staging phase now runs the same checks as the
upload REST path — `SegmentValidationUtils.validateTimeInterval`,
`validateUpsertSegmentPartitionMetadata`, and `checkStorageQuota` wired to the
controller's `StorageQuotaChecker` — so an over-quota table can no longer grow
through INSERT, and violations reject the whole statement before anything
reaches deep store or IdealState. If `addNewSegment` fails after the deep-store
copy, the orphan tar is deleted. Fuller reuse of ZKOperator (crypter,
lineage/refresh handling) is a follow-up; those paths don't apply to the v1
append-only ROW flow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
