xiangfu0 commented on code in PR #17939:
URL: https://github.com/apache/pinot/pull/17939#discussion_r4120685077
##########
pinot-core/src/main/java/org/apache/pinot/core/query/executor/sql/SqlQueryExecutor.java:
##########
@@ -112,6 +184,284 @@ public BrokerResponse
executeDMLStatement(SqlNodeAndOptions sqlNodeAndOptions,
return result;
}
+ private InsertResult executePushInsert(DataManipulationStatement statement,
+ @Nullable Map<String, String> headers)
+ throws Exception {
+ if (!(statement instanceof InsertIntoValues)) {
+ throw new IllegalStateException("PUSH execution type requires
InsertIntoValues statement");
+ }
+ InsertIntoValues insertStmt = (InsertIntoValues) statement;
+
+ InsertRequest request = buildInsertRequest(insertStmt);
+
+ // If a local executor is available (controller-side), call it directly
+ if (_insertExecutor != null) {
+ LOGGER.info("Executing push-based INSERT locally via InsertExecutor");
+ return _insertExecutor.execute(request);
+ }
+
+ // Otherwise, POST to controller /insert/execute (broker-side). Controller
requires
+ // ?tableName=... as a query parameter for table-scoped @Authorize
binding, so include it.
+ String controllerBaseUrl = getControllerUrl();
+ String tableName = request.getTableName();
+ if (tableName == null || tableName.isEmpty()) {
+ throw new RuntimeException("INSERT request missing tableName; cannot
route to controller");
+ }
+ String url = controllerBaseUrl + "/insert/execute?tableName="
+ + URLEncoder.encode(tableName, StandardCharsets.UTF_8);
+ String payload = JsonUtils.objectToString(request);
+
+ LOGGER.info("Submitting push-based INSERT to controller: {}", url);
+
+ // Forward caller-supplied headers (Authorization in particular). The
controller's
+ // /insert/execute is @Authenticate(AccessType.CREATE) +
@Authorize(action=EXECUTE_INSERT) and
+ // rejects unauthenticated requests with 401, so a null headers map fails
closed at the
+ // controller — but log a warning so misconfigured callers (forgot to
plumb headers through
+ // the broker dispatch) get a visible signal rather than a generic 401.
+ //
+ // Strip hop-by-hop and length-bearing headers from the inbound request
before forwarding —
+ // re-emitting "Content-Length" or "Transfer-Encoding" causes Apache
HttpClient to throw
+ // "Content-Length header already present" because sendJsonPostRequest
computes its own length
+ // from the new payload. The forwarded body has nothing to do with the
inbound body's framing.
+ // Use a case-insensitive map for HTTP headers per RFC 9110 §5.1. A plain
HashMap would let
+ // inbound "content-type" coexist with our explicit "Content-Type" if a
future filter change
+ // dropped a strip rule; case-insensitive comparison guarantees our
explicit headers win and
+ // there are no duplicate-cased entries on the wire.
+ Map<String, String> requestHeaders = new
TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ requestHeaders.put("Content-Type", "application/json");
+ requestHeaders.put("accept", "application/json");
+ // Note: the inbound map is single-valued per header name —
duplicate-cased inbound headers
+ // (e.g. two differently-cased Authorization values) were already
collapsed last-value-wins by
+ // the caller that built it. Acceptable here since only simple
single-valued headers matter.
+ if (headers != null && !headers.isEmpty()) {
+ for (Map.Entry<String, String> entry : headers.entrySet()) {
+ String name = entry.getKey();
+ if (name == null || isStrippedForwardedHeader(name)) {
+ continue;
+ }
+ requestHeaders.put(name, entry.getValue());
+ }
+ } else {
+ LOGGER.warn("executePushInsert: caller-supplied headers are null/empty;
the controller will "
+ + "reject this request if authentication is enabled. Plumb auth
headers through the "
+ + "broker dispatch path.");
+ }
+
+ String responseStr = HttpClient.wrapAndThrowHttpException(
+ HttpClient.getInstance().sendJsonPostRequest(URI.create(url), payload,
requestHeaders)).getResponse();
+ if (responseStr == null || responseStr.isEmpty()) {
+ throw new RuntimeException("Controller /insert/execute returned empty
response body for URL: " + url);
+ }
+ JsonNode responseJson;
+ try {
+ responseJson = JsonUtils.stringToJsonNode(responseStr);
+ } catch (Exception e) {
+ // Non-JSON response (HTML error page from a proxy, plaintext from a
misconfigured controller,
+ // etc.). Surface a clear diagnostic instead of letting Jackson's stack
trace bubble.
+ throw new RuntimeException("Controller /insert/execute returned non-JSON
response. URL=" + url
+ + ", first-200-chars=" + responseStr.substring(0, Math.min(200,
responseStr.length())), e);
+ }
+ return JsonUtils.jsonNodeToObject(responseJson, InsertResult.class);
+ }
+
+ /// Returns true for headers that must NOT be forwarded from the inbound
request to the
+ /// controller-bound POST. Length/framing-bearing headers (`Content-Length`,
+ /// `Transfer-Encoding`) cause Apache HttpClient to throw "Content-Length
header already
+ /// present" because the outbound client recomputes them from the new
payload. `Host` also
+ /// mismatches the controller URL. `Content-Type` and `Accept` are
explicitly set by
+ /// us to `application/json`; allowing inbound versions through would
clobber the JSON
+ /// contract.
+ private static boolean isStrippedForwardedHeader(String name) {
+ String lower = name.toLowerCase(Locale.ROOT);
+ return lower.equals("content-length")
+ || lower.equals("transfer-encoding")
+ || lower.equals("content-type")
+ || lower.equals("accept")
+ || lower.equals("host")
+ || lower.equals("connection")
+ || lower.equals("expect");
+ }
+
+ private static InsertRequest buildInsertRequest(InsertIntoValues insertStmt)
{
+ List<GenericRow> genericRows = new ArrayList<>();
+ List<String> columns = insertStmt.getColumns();
+ for (List<Object> rowValues : insertStmt.getRows()) {
+ GenericRow genericRow = new GenericRow();
+ for (int i = 0; i < columns.size(); i++) {
+ Object value = rowValues.get(i);
+ if (value == null) {
+ // Mark the field as null so that downstream segment generation uses
the default null
+ // value and records the null in the null bitmap, instead of
throwing on a raw null.
+ genericRow.putValue(columns.get(i), null);
+ genericRow.addNullValueField(columns.get(i));
+ } else {
+ genericRow.putValue(columns.get(i), value);
+ }
+ }
+ genericRows.add(genericRow);
+ }
+
+ TableType tableType = null;
+ String tableTypeStr = insertStmt.getTableType();
+ if (tableTypeStr != null) {
+ tableType = TableType.valueOf(tableTypeStr.toUpperCase(Locale.ROOT));
+ }
+
+ // Compute a stable payload hash for idempotency when requestId is set.
+ // Exclude control options (requestId, tableType) from the hash so the
payload hash
+ // only covers table + columns + values + non-control options.
+ Map<String, String> payloadOptions = null;
+ if (insertStmt.getOptions() != null) {
+ payloadOptions = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ payloadOptions.putAll(insertStmt.getOptions());
+ payloadOptions.remove("requestId");
+ payloadOptions.remove("tableType");
+ }
+ // Hash is only used for idempotency conflict detection paired with
requestId. When the client
+ // didn't supply a requestId the coordinator's idempotency path never
reads it, so skip the
+ // O(rows × columns) SHA-256 computation entirely.
+ String payloadHash = insertStmt.getRequestId() != null
+ ? computePayloadHash(insertStmt.getTableName(), columns,
insertStmt.getRows(), payloadOptions)
+ : null;
+
+ return new InsertRequest.Builder()
+ .setTableName(insertStmt.getTableName())
+ .setTableType(tableType)
+ .setInsertType(InsertType.ROW)
+ .setRows(genericRows)
+ .setRequestId(insertStmt.getRequestId())
+ .setPayloadHash(payloadHash)
+ .setOptions(insertStmt.getOptions())
Review Comment:
Fixed in adf8fb059e from both ends: the FILE executor now allowlists what
reaches Minion task configs (see the FileInsertExecutor thread), and the client
hash is gone entirely — the controller computes the payload hash from the
request itself, excluding the `requestId`/`tableType` control keys.
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/InsertIntoValuesClusterIntegrationTest.java:
##########
@@ -0,0 +1,404 @@
+/**
+ * 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.integration.tests;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.exception.HttpErrorStatusException;
+import org.apache.pinot.controller.ControllerConf;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.ingest.InsertConsistencyMode;
+import org.apache.pinot.spi.ingest.InsertType;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.apache.pinot.util.TestUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.*;
+
+
+/// Integration tests for push-based INSERT INTO functionality.
+///
+/// Tests the controller coordinator REST API for statement lifecycle
management,
+/// idempotency, hybrid table validation, abort, and list operations.
+///
+/// The ROW executor is registered with the coordinator at controller startup,
+/// so submitted inserts are expected to succeed (state = VISIBLE).
+///
+/// **Base class choice:** this test extends {@link
BaseClusterIntegrationTest} rather than
+/// {@link
org.apache.pinot.integration.tests.custom.CustomDataQueryClusterIntegrationTest}
+/// for three concrete reasons:
+/// - **Per-test table topology** — each test method creates its own OFFLINE
and/or REALTIME
+/// table inside the test body and tears it down at the end. This is
required because tests
+/// exercise hybrid table validation (which needs both _OFFLINE and
_REALTIME variants present),
+/// table-not-found rejection (needs a table to NOT exist), and idempotency
across distinct
+/// tables. The `CustomDataQueryClusterIntegrationTest` pattern of one
shared schema/table
+/// per test class cannot express these topologies.
+/// - **Coordinator state isolation** — each test asserts on
coordinator-internal state
+/// (active-statement counter, request-id reservation tombstones, manifest
GC). Sharing tables
+/// across tests would let one test's manifest leak into another's
listStatements assertions.
+/// - **Controller config override** — {@link #overrideControllerConf} sets
+/// `controller.insert.enabled=true`. `BaseClusterIntegrationTest` runs
setUp/tearDown
+/// per class so the flag binding is stable; cluster restart per method is
not needed.
+public class InsertIntoValuesClusterIntegrationTest extends
BaseClusterIntegrationTest {
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(InsertIntoValuesClusterIntegrationTest.class);
+
+ @Override
+ protected void overrideControllerConf(Map<String, Object> properties) {
+ /// Enable the push-based INSERT INTO feature flag. The flag defaults to
false in production
+ /// so operators must opt in; integration tests need it on to exercise the
coordinator.
+ properties.put(ControllerConf.INSERT_ENABLED, true);
+ }
+
+ @BeforeClass
+ public void setUp()
+ throws Exception {
+ TestUtils.ensureDirectoriesExistAndEmpty(_tempDir);
+ startZk();
+ startController();
+ startBroker();
+ startServer();
+ }
+
+ @AfterClass
+ public void tearDown() {
+ try {
+ stopServer();
+ stopBroker();
+ stopController();
+ stopZk();
+ } finally {
+ FileUtils.deleteQuietly(_tempDir);
+ }
+ }
+
+ private void createOfflineTable(String tableName)
+ throws Exception {
+ Schema schema = new Schema.SchemaBuilder()
+ .setSchemaName(tableName)
+ .addSingleValueDimension("id", FieldSpec.DataType.INT)
+ .addSingleValueDimension("name", FieldSpec.DataType.STRING)
+ .addMetric("score", FieldSpec.DataType.FLOAT)
+ .build();
+ addSchema(schema);
+
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+ .setTableName(tableName)
+ .build();
+ sendPostRequest(
+ _controllerRequestURLBuilder.forTableCreate(),
+ tableConfig.toString(),
+ BasicAuthTestUtils.AUTH_HEADER);
+ }
+
+ private String buildInsertRequestJson(String tableName, TableType tableType,
String requestId,
+ String payloadHash)
+ throws Exception {
+ Map<String, Object> request = new HashMap<>();
+ request.put("tableName", tableName);
+ request.put("insertType", InsertType.ROW.name());
+ request.put("consistencyMode",
InsertConsistencyMode.WAIT_FOR_ACCEPT.name());
+
+ if (tableType != null) {
+ request.put("tableType", tableType.name());
+ }
+ if (requestId != null) {
+ request.put("requestId", requestId);
+ }
+ if (payloadHash != null) {
+ request.put("payloadHash", payloadHash);
+ }
+
+ /// Minimal rows
+ java.util.List<Map<String, Object>> rowList = new java.util.ArrayList<>();
+ Map<String, Object> row = new HashMap<>();
+ Map<String, Object> fieldMap = new HashMap<>();
+ fieldMap.put("id", 1);
+ fieldMap.put("name", "test");
+ fieldMap.put("score", 90.0);
+ row.put("fieldToValueMap", fieldMap);
+ rowList.add(row);
+ request.put("rows", rowList);
+
+ return JsonUtils.objectToString(request);
+ }
+
+ private JsonNode postInsertRequest(String payload)
+ throws Exception {
+ /// The /insert/execute endpoint requires ?tableName=<table> as a query
param to bind the
+ /// table-scoped @Authorize check. Extract the value from the payload so
callers don't need to
+ /// pass it twice.
+ String tableName =
JsonUtils.stringToJsonNode(payload).get("tableName").asText();
+ String url = getControllerBaseApiUrl() + "/insert/execute?tableName="
+ + URLEncoder.encode(tableName, StandardCharsets.UTF_8);
+ String response = sendPostRequest(url, payload,
+ Collections.singletonMap("accept", "application/json"));
+ return JsonUtils.stringToJsonNode(response);
+ }
+
+ private JsonNode getInsertStatus(String statementId, String
tableNameWithType)
+ throws Exception {
+ String url = getControllerBaseApiUrl() + "/insert/status/" + statementId
+ + "?tableName=" + tableNameWithType;
+ String response = sendGetRequest(url);
+ return JsonUtils.stringToJsonNode(response);
+ }
+
+ private JsonNode postInsertAbort(String statementId, String
tableNameWithType)
+ throws Exception {
+ String url = getControllerBaseApiUrl() + "/insert/abort/" + statementId
+ + "?tableName=" + tableNameWithType;
+ String response = sendPostRequest(url, null,
+ Collections.singletonMap("accept", "application/json"));
+ return JsonUtils.stringToJsonNode(response);
+ }
+
+ private JsonNode getInsertList(String tableNameWithType)
+ throws Exception {
+ String url = getControllerBaseApiUrl() + "/insert/list?tableName=" +
tableNameWithType;
+ String response = sendGetRequest(url);
+ return JsonUtils.stringToJsonNode(response);
+ }
+
+ /// ---- Test: Submit insert and verify coordinator processes it ----
+
+ @Test
+ public void testInsertIntoOfflineTableReturnsResult()
+ throws Exception {
+ String tableName = "insertOfflineResult";
+ createOfflineTable(tableName);
+
+ String payload = buildInsertRequestJson(tableName, null, null, null);
+
+ JsonNode result = postInsertRequest(payload);
+ LOGGER.info("Insert result: {}", result);
+
+ assertNotNull(result.get("statementId"), "statementId should be present");
+ String statementId = result.get("statementId").asText();
+ assertFalse(statementId.isEmpty(), "statementId should not be empty");
+
+ /// With ROW executor registered, the insert should reach VISIBLE
synchronously.
+ String state = result.get("state").asText();
+ assertEquals(state, "VISIBLE", "ROW executor should return VISIBLE
synchronously, got: " + state);
+ }
+
+ /// ---- Test: Explicit OFFLINE type targeting works for single-type table
----
+
+ @Test
+ public void testInsertWithExplicitOfflineType()
+ throws Exception {
+ String tableName = "insertExplicitOffline";
+ createOfflineTable(tableName);
+
+ /// Submit with explicit tableType=OFFLINE on a table that only has OFFLINE
+ String payload = buildInsertRequestJson(tableName, TableType.OFFLINE,
null, null);
+ JsonNode result = postInsertRequest(payload);
+ LOGGER.info("Explicit OFFLINE type result: {}", result);
+ /// Assert the request resolves and reaches VISIBLE — pins the
explicit-type resolution path.
+ assertEquals(result.get("state").asText(), "VISIBLE",
+ "Explicit OFFLINE type on an OFFLINE-only table should resolve and
produce VISIBLE; got: " + result);
+ }
+
+ /// ---- Test: Table does not exist ----
+
+ @Test
+ public void testInsertNonExistentTable()
+ throws Exception {
+ String payload = buildInsertRequestJson("nonExistentTable99", null, null,
null);
+
+ JsonNode result = postInsertRequest(payload);
+ LOGGER.info("Non-existent table result: {}", result);
+ assertEquals(result.get("state").asText(), "REJECTED");
+ assertEquals(result.get("errorCode").asText(), "TABLE_RESOLUTION_ERROR");
+ }
+
+ /// ---- Test: Insert with wrong explicit table type ----
+
+ @Test
+ public void testInsertWithWrongTableType()
+ throws Exception {
+ String tableName = "insertWrongType";
+ createOfflineTable(tableName);
+
+ /// Try to insert with REALTIME type into an OFFLINE-only table
+ String payload = buildInsertRequestJson(tableName, TableType.REALTIME,
null, null);
+ JsonNode result = postInsertRequest(payload);
+ LOGGER.info("Wrong type result: {}", result);
+ assertEquals(result.get("state").asText(), "REJECTED");
+ assertEquals(result.get("errorCode").asText(), "TABLE_RESOLUTION_ERROR");
+ }
+
+ /// ---- Test: Insert with explicit table type suffix in name ----
+
+ @Test
+ public void testInsertWithExplicitTableTypeSuffix()
+ throws Exception {
+ String tableName = "insertSuffix";
+ createOfflineTable(tableName);
+
+ /// Use explicit _OFFLINE suffix in table name
+ String payload = buildInsertRequestJson(tableName + "_OFFLINE", null,
null, null);
+ JsonNode result = postInsertRequest(payload);
+ LOGGER.info("Explicit suffix result: {}", result);
+ String state = result.get("state").asText();
+ assertEquals(state, "VISIBLE", "ROW executor should return VISIBLE
synchronously, got: " + state);
+ }
+
+ /// ---- Test: Idempotency — same requestId + payloadHash should return the
+ /// same statementId and not re-execute ----
+
+ @Test
+ public void testInsertIdempotency()
+ throws Exception {
+ String tableName = "insertIdemp";
+ createOfflineTable(tableName);
+
+ String requestId = "idemp-001";
+ String payloadHash = "hash-abc";
+
+ String payload = buildInsertRequestJson(tableName, null, requestId,
payloadHash);
+
+ /// First submit — succeeds with executor registered
+ JsonNode result1 = postInsertRequest(payload);
+ LOGGER.info("First submit: {}", result1);
+ String state1 = result1.get("state").asText();
+ assertFalse("REJECTED".equals(state1) && result1.has("errorCode")
+ && "NO_EXECUTOR".equals(result1.get("errorCode").asText()),
+ "Executor should be registered; should not get NO_EXECUTOR");
+ String statementId1 = result1.get("statementId").asText();
+
+ /// Second submit with same requestId + payloadHash — should be idempotent
+ JsonNode result2 = postInsertRequest(payload);
+ LOGGER.info("Second submit: {}", result2);
+ assertEquals(result2.get("statementId").asText(), statementId1,
+ "Idempotent retry should return same statementId");
+ }
+
+ /// ---- Test: Status API returns NOT_FOUND for unknown statement ----
+
+ @Test
+ public void testInsertStatusNotFound()
+ throws Exception {
+ String tableName = "insertStatusNotFound";
+ createOfflineTable(tableName);
+
+ /// The resource maps any state=REJECTED coordinator response (including
NOT_FOUND) to HTTP 404
+ /// to honor the contract that REJECTED has no manifest persisted.
ControllerTest.sendGetRequest
+ /// wraps HttpErrorStatusException in an IOException on 4xx, so we catch
IOException and unwrap
+ /// to check the status code and error message.
+ try {
+ getInsertStatus("unknown-stmt-id", tableName + "_OFFLINE");
+ fail("Expected HTTP 404 for unknown statementId");
+ } catch (IOException e) {
+ Throwable cause = e.getCause();
+ assertTrue(cause instanceof HttpErrorStatusException,
+ "Expected cause to be HttpErrorStatusException, got: " + cause);
+ HttpErrorStatusException httpEx = (HttpErrorStatusException) cause;
+ assertEquals(httpEx.getStatusCode(), 404);
+ assertTrue(httpEx.getMessage().contains("errorCode=NOT_FOUND"),
+ "Expected message to mention errorCode=NOT_FOUND, got: " +
httpEx.getMessage());
+ }
+ }
+
+ /// ---- Test: Abort returns NOT_FOUND for unknown statement ----
+
+ @Test
+ public void testInsertAbortNotFound()
+ throws Exception {
+ String tableName = "insertAbortNotFound";
+ createOfflineTable(tableName);
+
+ JsonNode abortResult = postInsertAbort("unknown-stmt-id", tableName +
"_OFFLINE");
+ LOGGER.info("Abort not found: {}", abortResult);
+ assertEquals(abortResult.get("errorCode").asText(), "NOT_FOUND");
+ }
+
+ /// ---- Test: List returns empty array for table with no statements ----
+
+ @Test
+ public void testInsertListEmpty()
+ throws Exception {
+ String tableName = "insertListEmpty";
+ createOfflineTable(tableName);
+
+ JsonNode list = getInsertList(tableName + "_OFFLINE");
+ LOGGER.info("Empty list: {}", list);
+ assertTrue(list.isArray(), "List should return an array");
+ assertEquals(list.size(), 0, "Should have 0 statements");
+ }
+
+ /// ---- Test: SQL parsing of INSERT INTO VALUES through broker ----
+
+ @Test
+ public void testInsertIntoValuesSqlParsing()
Review Comment:
Partially addressed in adf8fb059e: added a typed round-trip IT through the
broker (negative INT, BYTES via X'..' literal, BIG_DECIMAL precision, negative
DOUBLE) covering the riskiest wire-fidelity path. The
REALTIME-uploaded-next-to-consuming, full-upsert partition assignment,
multi-controller, and hybrid-tableType scenarios are real gaps — proposing to
track them as a follow-up IT PR (they need meaningfully different cluster
topologies than this test class); the multi-controller sweep race specifically
is mitigated by the new lead-controller gating.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/ingest/InsertRequest.java:
##########
@@ -0,0 +1,251 @@
+/**
+ * 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.spi.ingest;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+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 org.apache.pinot.spi.annotations.InterfaceStability;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+
+
+/// Immutable data transfer object representing an INSERT INTO request.
+///
+/// Carries all information needed to execute a single INSERT statement,
including the data
+/// payload (rows or file URI), target table, idempotency keys, and
consistency preferences.
+///
+/// Instances are created via the {@link Builder} or deserialized from JSON.
Immutable and
+/// therefore thread-safe.
+@JsonIgnoreProperties(ignoreUnknown = true)
[email protected]
+public class InsertRequest {
+ private final String _statementId;
+ private final String _requestId;
+ private final String _payloadHash;
+ private final String _tableName;
+ private final TableType _tableType;
+ private final InsertType _insertType;
+ private final List<GenericRow> _rows;
+ private final String _fileUri;
+ private final Map<String, String> _options;
+ private final InsertConsistencyMode _consistencyMode;
+
+ @JsonCreator
+ public InsertRequest(
+ @JsonProperty("statementId") String statementId,
+ @JsonProperty("requestId") String requestId,
+ @JsonProperty("payloadHash") String payloadHash,
+ @JsonProperty("tableName") String tableName,
+ @JsonProperty("tableType") TableType tableType,
+ @JsonProperty("insertType") InsertType insertType,
+ @JsonProperty("rows") List<GenericRow> rows,
+ @JsonProperty("fileUri") String fileUri,
+ @JsonProperty("options") Map<String, String> options,
+ @JsonProperty("consistencyMode") InsertConsistencyMode consistencyMode) {
+ /// Mirror Builder.build()'s invariants on the wire-deserialized path so a
malformed JSON body
+ /// is rejected at construction with a clear message rather than silently
propagating a null
+ /// through the coordinator and surfacing as an NPE later. Wording is
intentionally identical
+ /// to the Builder.build() throws so log-greps catch both paths.
+ if (tableName == null || tableName.isEmpty()) {
+ throw new IllegalArgumentException("tableName is required for
InsertRequest");
+ }
+ if (insertType == null) {
+ throw new IllegalArgumentException("insertType is required for
InsertRequest");
+ }
+ if (insertType == InsertType.FILE && (fileUri == null ||
fileUri.isEmpty())) {
+ throw new IllegalArgumentException("fileUri is required for FILE
insert");
+ }
+ _statementId = statementId != null ? statementId :
UUID.randomUUID().toString();
Review Comment:
Fixed in adf8fb059e: `/insert/execute` now ignores any client-supplied
statementId and always substitutes a server-generated UUID
(`withServerGeneratedStatementId()`); idempotent retries correlate via
requestId, and the assigned statementId comes back in the InsertResult. As
defense in depth the wire constructor also validates the field against
`ID_PATTERN`, so even direct SPI users can't smuggle a path-breaking value.
--
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]
