wasabii commented on code in PR #4922:
URL: https://github.com/apache/calcite/pull/4922#discussion_r3336264774
##########
core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableTableModify.java:
##########
@@ -80,100 +88,394 @@ public EnumerableTableModify(RelOptCluster cluster,
RelTraitSet traits,
final BlockBuilder builder = new BlockBuilder();
final Result result =
implementor.visitChild(this, 0, (EnumerableRel) getInput(), pref);
- Expression childExp =
- builder.append(
- "child", result.block);
+
+ // Enumerable produced by the input relational expression.
+ final Expression sourceExp =
+ builder.append("source", result.block);
+
+ // Variable that will hold the table's mutable backing collection.
final ParameterExpression collectionParameter =
Expressions.parameter(Collection.class,
builder.newName("collection"));
+
+ // Expression that yields the ModifiableTable instance at runtime.
final Expression expression = table.getExpression(ModifiableTable.class);
requireNonNull(expression, "expression"); // TODO: user error in validator
checkArgument(
ModifiableTable.class.isAssignableFrom(
Types.toClass(expression.getType())),
"not assignable from type %s", expression.getType());
+
+ // collection = table.getModifiableCollection()
builder.add(
Expressions.declare(
Modifier.FINAL,
collectionParameter,
Expressions.call(
expression,
- BuiltInMethod.MODIFIABLE_TABLE_GET_MODIFIABLE_COLLECTION
- .method)));
+
BuiltInMethod.MODIFIABLE_TABLE_GET_MODIFIABLE_COLLECTION.method)));
+
+ // Physical row representation of this TableModify node's output
+ // (a single ROWCOUNT field).
+ final PhysType physType =
+ PhysTypeImpl.of(
+ implementor.getTypeFactory(),
+ getRowType(),
+ pref == Prefer.ARRAY ? JavaRowFormat.ARRAY : JavaRowFormat.SCALAR);
+
+ switch (getOperation()) {
+ case INSERT:
+ return implementInsert(implementor, result, builder, sourceExp,
collectionParameter, physType);
+ case UPDATE:
+ return implementUpdate(implementor, builder, sourceExp,
collectionParameter, physType);
+ case DELETE:
+ return implementDelete(implementor, builder, sourceExp,
collectionParameter, physType);
+ default:
+ throw new AssertionError("unsupported operation: " + getOperation());
+ }
+ }
+
+ /** Generates code for an UPDATE statement.
+ *
+ * <p>Applies updates to matching rows in the backing collection and returns
+ * the number of updated rows as a single-element enumerable.
+ *
+ * @param implementor code-generation context
+ * @param builder block under construction
+ * @param sourceExp enumerable of source rows produced by the input
+ * @param collectionParameter the modifiable backing collection of the table
+ * @param physType physical type of this node's output row
+ */
+ private Result implementUpdate(
+ EnumerableRelImplementor implementor,
+ BlockBuilder builder,
+ Expression sourceExp,
+ ParameterExpression collectionParameter,
+ PhysType physType) {
+ final List<String> updateCols = requireNonNull(getUpdateColumnList());
+ final List<RelDataTypeField> tableFields =
table.getRowType().getFieldList();
+ final int tableFieldCount = tableFields.size();
+
+ // Child row layout for UPDATE:
+ // [originalField_0, ..., originalField_N-1, newValue_0, ..., newValue_M-1]
+ // where N = tableFieldCount and M = updateCols.size().
+
+ // Resolve each SET-column name to its 0-based position in the table row.
+ final int[] updateColumnIndices = new int[updateCols.size()];
+ for (int i = 0; i < updateCols.size(); i++) {
+ final String colName = updateCols.get(i);
+ int found = -1;
+ for (int j = 0; j < tableFields.size(); j++) {
+ if (tableFields.get(j).getName().equals(colName)) {
+ found = j;
+ break;
+ }
+ }
+ if (found < 0) {
+ throw new AssertionError("column '" + colName + "' not found in
table");
+ }
+ updateColumnIndices[i] = found;
+ }
+
+ // Generate code that applies one-to-one update consumption:
+ // each source row updates at most one matching sink row.
+ final Expression updateCountExp =
+ builder.append(
+ "updateCount",
+ Expressions.call(
+ EnumerableTableModify.class,
+ "applyUpdateOneToOne",
+ // Source rows are produced by the child relational expression.
+ sourceExp,
+ // Sink is the table's mutable backing collection.
+ Expressions.convert_(collectionParameter, List.class),
+ // Number of original-row fields in each source payload.
+ Expressions.constant(tableFieldCount),
+ // Table column positions to overwrite from trailing source
values.
+ Expressions.constant(updateColumnIndices)));
+
+ // Return the number of updated rows as the single output row.
+ builder.add(
+ Expressions.return_(
+ null,
+ Expressions.call(
+ BuiltInMethod.SINGLETON_ENUMERABLE.method,
+ Expressions.convert_(
+ updateCountExp,
+ long.class))));
+
+ return implementor.result(physType, builder.toBlock());
+ }
+
+ /**
+ * Applies UPDATE with one-to-one, first-match consumption semantics.
+ *
+ * <p>Each source row contributes one replacement row keyed by the original
+ * row content. As sink rows are scanned in order, the first matching row for
+ * each queued source update is replaced and consumed, so duplicate keys
update
+ * only as many rows as appear in the source.
+ */
+ @SuppressWarnings({"unchecked"})
+ public static long applyUpdateOneToOne(Enumerable<?> source, List<?> sink,
+ int tableFieldCount, int[] updateColumnIndices) {
+ final Map<List<Object>, Deque<Object[]>> updatesByKey = new HashMap<>();
+ try (Enumerator<?> e = source.enumerator()) {
+ while (e.moveNext()) {
+ final Object[] sourceRow = (Object[]) e.current();
+ final List<Object> key = Arrays.asList(Arrays.copyOf(sourceRow,
tableFieldCount));
+ final Object[] newRow = applyUpdate(sourceRow, tableFieldCount,
updateColumnIndices);
+ updatesByKey.computeIfAbsent(key, k -> new
ArrayDeque<>()).addLast(newRow);
+ }
+ }
+
+ long updateCount = 0;
+ final ListIterator<Object[]> it = ((List<Object[]>) sink).listIterator();
+ while (it.hasNext()) {
+ final Object[] current = it.next();
+ final List<Object> key = Arrays.asList(current);
+ final Deque<Object[]> pending = updatesByKey.get(key);
+ if (pending == null || pending.isEmpty()) {
+ continue;
+ }
+ it.set(pending.removeFirst());
+ updateCount++;
+ if (pending.isEmpty()) {
+ updatesByKey.remove(key);
+ }
+ }
+ return updateCount;
+ }
+
+ /** Generates code for a DELETE statement.
+ *
+ * <p>The source produces every row that matches the WHERE clause. Those rows
+ * are removed from the backing collection and the number of deleted rows is
+ * returned as a single-element enumerable.
+ *
+ * @param implementor code-generation context
+ * @param builder block under construction
+ * @param sourceExp enumerable of rows to delete — every row matched
+ * by the WHERE clause, as produced by the input
+ * relational expression
+ * @param collectionParameter the modifiable backing collection of the table
+ * @param physType physical type of this node's output row
+ */
+ private Result implementDelete(
+ EnumerableRelImplementor implementor,
+ BlockBuilder builder,
+ Expression sourceExp,
+ ParameterExpression collectionParameter,
+ PhysType physType) {
+
+ // Snapshot the collection size before the delete so we can compute the
+ // number of rows removed as (sizeBefore - sizeAfter).
final Expression countParameter =
builder.append(
"count",
Expressions.call(collectionParameter, "size"),
false);
- Expression convertedChildExp;
+
+ final String deleteMethodName =
+ EnumerableTableScan.deduceFormat(table) == JavaRowFormat.ARRAY
+ ? "applyDeleteArrayRows"
+ : "applyDeleteScalarRows";
+
+ // Remove every row in sourceExp from the backing collection.
+ builder.add(
+ Expressions.statement(
+ Expressions.call(
+ EnumerableTableModify.class,
+ deleteMethodName,
+ sourceExp,
+ collectionParameter)));
+
+ // Snapshot the size again and return (sizeBefore - sizeAfter) as the
delete count.
+ final Expression updatedCountParameter =
+ builder.append(
+ "updatedCount",
+ Expressions.call(collectionParameter, "size"),
+ false);
+
+ builder.add(
+ Expressions.return_(
+ null,
+ Expressions.call(
+ BuiltInMethod.SINGLETON_ENUMERABLE.method,
+ Expressions.convert_(
+ Expressions.subtract(countParameter,
updatedCountParameter),
+ long.class))));
+
+ return implementor.result(physType, builder.toBlock());
+ }
+
+ /** Generates code for an INSERT statement.
+ *
+ * <p>All rows produced by the source are added to the backing collection and
+ * the number of inserted rows is returned as a single-element enumerable.
+ *
+ * @param implementor code-generation context
+ * @param result compiled result of the input relational expression
+ * @param builder block under construction
+ * @param sourceExp enumerable of rows to insert — the output of the input
+ * relational expression (VALUES, SELECT, etc.) with any
+ * upstream filtering or projection already applied
+ * @param collectionParameter the modifiable backing collection of the table
+ * @param physType physical type of this node's output row
+ */
+ private Result implementInsert(
+ EnumerableRelImplementor implementor,
+ Result result,
+ BlockBuilder builder,
+ Expression sourceExp,
+ ParameterExpression collectionParameter,
+ PhysType physType) {
+
+ // Snapshot the collection size before the insert so we can compute the
+ // number of rows added as (sizeAfter - sizeBefore).
+ final Expression countParameter =
+ builder.append(
+ "count",
+ Expressions.call(collectionParameter, "size"),
+ false);
+
+ // insertExp is the enumerable that will actually be streamed into the
+ // collection. When the source row type matches the table's row type
+ // exactly it is the same as sourceExp; otherwise it is sourceExp wrapped
+ // in a field-by-field cast projection so that every value lands in the
+ // Java type the table's backing collection expects.
+ Expression insertExp;
if (!getInput().getRowType().equals(getRowType())) {
+ // The source row type doesn't match the table's row type (e.g. types
+ // differ in nullability or precision), so wrap the source in a
projection
+ // that casts each field to the exact Java type the table expects.
final JavaTypeFactory typeFactory =
(JavaTypeFactory) getCluster().getTypeFactory();
final JavaRowFormat format = EnumerableTableScan.deduceFormat(table);
- PhysType physType =
+ PhysType tablePhysType =
PhysTypeImpl.of(typeFactory, table.getRowType(), format);
+ // One cast expression per field: sourceField -> tableFieldType.
List<Expression> expressionList = new ArrayList<>();
- final PhysType childPhysType = result.physType;
+ final PhysType sourcePhysType = result.physType;
final ParameterExpression o_ =
- Expressions.parameter(childPhysType.getJavaRowType(), "o");
- final int fieldCount =
- childPhysType.getRowType().getFieldCount();
+ Expressions.parameter(sourcePhysType.getJavaRowType(), "o");
+ final int fieldCount = sourcePhysType.getRowType().getFieldCount();
for (int i = 0; i < fieldCount; i++) {
expressionList.add(
- childPhysType.fieldReference(o_, i, physType.getJavaFieldType(i)));
+ sourcePhysType.fieldReference(o_, i,
tablePhysType.getJavaFieldType(i)));
}
- convertedChildExp =
+ // insertExp = sourceExp.select(o -> new TableRow(cast(o.f0),
cast(o.f1), ...))
+ insertExp =
builder.append(
- "convertedChild",
+ "insertRows",
Expressions.call(
- childExp,
+ sourceExp,
BuiltInMethod.SELECT.method,
- Expressions.lambda(
- physType.record(expressionList), o_)));
+ Expressions.lambda(tablePhysType.record(expressionList),
o_)));
} else {
- convertedChildExp = childExp;
- }
- final Method method;
- switch (getOperation()) {
- case INSERT:
- method = BuiltInMethod.INTO.method;
- break;
- case DELETE:
- method = BuiltInMethod.REMOVE_ALL.method;
- break;
- default:
- throw new AssertionError(getOperation());
+ insertExp = sourceExp;
}
+
+ // Stream all rows from insertExp into the backing collection.
builder.add(
Expressions.statement(
Expressions.call(
- convertedChildExp, method, collectionParameter)));
+ insertExp, BuiltInMethod.INTO.method, collectionParameter)));
+
+ // Snapshot the size again and return (sizeAfter - sizeBefore) as the
insert count.
final Expression updatedCountParameter =
builder.append(
"updatedCount",
Expressions.call(collectionParameter, "size"),
false);
+
builder.add(
Expressions.return_(
null,
Expressions.call(
BuiltInMethod.SINGLETON_ENUMERABLE.method,
Expressions.convert_(
- Expressions.condition(
- Expressions.greaterThanOrEqual(
- updatedCountParameter, countParameter),
- Expressions.subtract(
- updatedCountParameter, countParameter),
- Expressions.subtract(
- countParameter, updatedCountParameter)),
+ Expressions.subtract(updatedCountParameter,
countParameter),
long.class))));
- final PhysType physType =
- PhysTypeImpl.of(
- implementor.getTypeFactory(),
- getRowType(),
- pref == Prefer.ARRAY
- ? JavaRowFormat.ARRAY : JavaRowFormat.SCALAR);
+
return implementor.result(physType, builder.toBlock());
}
+ /**
+ * Removes from {@code sink} one occurrence per occurrence in {@code source}
+ * when rows are represented as {@code Object[]}.
+ */
+ @SuppressWarnings({"unchecked"})
+ public static void applyDeleteArrayRows(Enumerable<?> source, Collection<?>
sink) {
+ final Collection<Object[]> sinkRows = (Collection<Object[]>) sink;
+ final List<Object[]> sourceRows = new ArrayList<>();
+ try (Enumerator<?> e = source.enumerator()) {
+ while (e.moveNext()) {
+ sourceRows.add((Object[]) e.current());
+ }
+ }
+ // Drain source first, then mutate sink, to avoid iterator interference
+ // when source and sink share the same backing collection.
+ for (Object[] row : sourceRows) {
+ removeFirstArrayMatch(sinkRows, row);
+ }
+ }
+
+ /**
+ * Removes from {@code sink} one occurrence per occurrence in {@code source}
+ * for scalar/custom row representations.
+ */
+ @SuppressWarnings({"unchecked"})
+ public static void applyDeleteScalarRows(Enumerable<?> source, Collection<?>
sink) {
+ final Collection<Object> sinkRows = (Collection<Object>) sink;
+ final List<Object> sourceRows = new ArrayList<>();
+ try (Enumerator<?> e = source.enumerator()) {
+ while (e.moveNext()) {
+ sourceRows.add(e.current());
+ }
+ }
+ // Drain source first, then mutate sink, to avoid iterator interference
Review Comment:
Comment rewritten from scratch.
--
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]
