Merge remote-tracking branch 'origin/master' into ignite-26212

Author: anton-vinogradov

docs/_docs/SQL/schemas.adoc

@@ -92,3 +92,52 @@ If you do not use this parameter, the cache name is defined in the following for
 ....
 SQL_<SCHEMA_NAME>_<TABLE_NAME>
 ....
+
+
+== Validating Data Against the Schema
+
+Ignite can verify that values inserted into a table match the column types and constraints declared in its SQL schema. By default, Ignite performs these checks only for indexed columns. To validate all columns, enable the server-side check on every node before startup:
+
+[tabs]
+--
+tab:Java[]
+[source,java]
+----
+SqlConfiguration sqlCfg = new SqlConfiguration()
+    .setValidationEnabled(true);
+
+IgniteConfiguration cfg = new IgniteConfiguration()
+    .setSqlConfiguration(sqlCfg);
+----
+
+tab:XML[]
+[source,xml]
+----
+<property name="sqlConfiguration">
+    <bean class="org.apache.ignite.configuration.SqlConfiguration">
+        <property name="validationEnabled" value="true"/>
+    </bean>
+</property>
+----
+--
+
+When validation is enabled, Ignite runs extra checks for each SQL `INSERT`, `MERGE`, and `UPDATE` statement as well as cache API calls that modify SQL-managed tables and rejects data that breaks the schema. Violations raise an `IgniteSQLException` for SQL clients (JDBC, ODBC, REST), while key-value API callers get a `CacheException` whose root cause is that same `IgniteSQLException`.
+In both cases the application can handle the failure and the incorrect values are not stored. SQL DML already attempts to coerce values to the declared column types, so type mismatches typically appear when you write data through the cache API or binary objects.
+
+[source,java]
+----
+// CREATE TABLE Person (id INT PRIMARY KEY, age INT);
+
+IgniteCache<Integer, BinaryObject> cache = ignite.cache("SQL_PUBLIC_PERSON").withKeepBinary();
+
+BinaryObject invalidPerson = ignite.binary().builder("Person")
+    .setField("id", 2)
+    .setField("age", "forty-two") // String instead of INT column
+    .build();
+
+cache.put(2, invalidPerson); // CacheException wrapping IgniteSQLException when validation is enabled
+----
+
+With the default configuration (validation disabled for non-indexed columns), the `put` above succeeds even though the stored `BinaryObject` does not match the SQL schema. Turning on validation causes Ignite to reject the update, protecting the table from malformed records.
+
+Enable the option when you need stronger guarantees that dynamic or user-provided data cannot break the table definition, and keep it disabled when the overhead of additional checks outweighs the risk of incorrect data.

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java

@@ -27,6 +27,7 @@
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 import org.apache.calcite.rel.RelCollation;
+import org.apache.calcite.rel.RelFieldCollation;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.core.Intersect;
 import org.apache.calcite.rel.core.JoinRelType;
@@ -37,6 +38,7 @@
 import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.mapping.IntPair;
 import org.apache.ignite.internal.processors.failure.FailureProcessor;
 import org.apache.ignite.internal.processors.query.QueryUtils;
 import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler.RowFactory;
@@ -116,8 +118,6 @@
 import org.apache.ignite.internal.util.typedef.F;
 
 import static org.apache.calcite.rel.RelDistribution.Type.HASH_DISTRIBUTED;
-import static org.apache.calcite.sql.SqlKind.IS_DISTINCT_FROM;
-import static org.apache.calcite.sql.SqlKind.IS_NOT_DISTINCT_FROM;
 import static org.apache.ignite.internal.processors.query.calcite.util.TypeUtils.combinedRowType;
 
 /**
@@ -298,12 +298,40 @@ public LogicalRelImplementor(
         RelDataType rightType = rel.getRight().getRowType();
         JoinRelType joinType = rel.getJoinType();
 
-        int pairsCnt = rel.analyzeCondition().pairs().size();
+        List<IntPair> joinPairs = rel.analyzeCondition().pairs();
+        int pairsCnt = joinPairs.size();
+
+        List<RelFieldCollation> leftCollations = rel.leftCollation().getFieldCollations();
+        List<RelFieldCollation> rightCollations = rel.rightCollation().getFieldCollations();
+
+        ImmutableBitSet allowNulls = rel.allowNulls();
+        ImmutableBitSet.Builder collsAllowNullsBuilder = ImmutableBitSet.builder();
+        int lastCollField = -1;
+
+        for (int c = 0; c < Math.min(leftCollations.size(), rightCollations.size()); ++c) {
+            RelFieldCollation leftColl = leftCollations.get(c);
+            RelFieldCollation rightColl = rightCollations.get(c);
+            collsAllowNullsBuilder.set(c);
+
+            for (int p = 0; p < pairsCnt; ++p) {
+                IntPair pair = joinPairs.get(p);
+
+                if (pair.source == leftColl.getFieldIndex() && pair.target == rightColl.getFieldIndex()) {
+                    lastCollField = c;
+
+                    if (!allowNulls.get(p)) {
+                        collsAllowNullsBuilder.clear(c);
+
+                        break;
+                    }
+                }
+            }
+        }
 
         Comparator<Row> comp = expressionFactory.comparator(
-            rel.leftCollation().getFieldCollations().subList(0, pairsCnt),
-            rel.rightCollation().getFieldCollations().subList(0, pairsCnt),
-            rel.getCondition().getKind() == IS_NOT_DISTINCT_FROM || rel.getCondition().getKind() == IS_DISTINCT_FROM
+            leftCollations.subList(0, lastCollField + 1),
+            rightCollations.subList(0, lastCollField + 1),
+            collsAllowNullsBuilder.build()
         );
 
         Node<Row> node = MergeJoinNode.create(ctx, outType, leftType, rightType, joinType, comp, hasExchange(rel));

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/ExpressionFactory.java

@@ -23,13 +23,13 @@
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
-
 import org.apache.calcite.rel.RelCollation;
 import org.apache.calcite.rel.RelFieldCollation;
 import org.apache.calcite.rel.core.AggregateCall;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.util.ImmutableBitSet;
 import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AccumulatorWrapper;
 import org.apache.ignite.internal.processors.query.calcite.exec.exp.agg.AggregateType;
 import org.apache.ignite.internal.processors.query.calcite.prepare.bounds.SearchBounds;
@@ -60,11 +60,14 @@ Supplier<List<AccumulatorWrapper<Row>>> accumulatorsFactory(
      *
      * @param left Collations of left row.
      * @param right Collations of right row.
-     * @param nullsEqual If {@code true}, nulls are considered equal. Usually, NULL <> NULL in SQL. So, the value should
-     *                   be {@code false}. Except cases with IS DISTINCT / IS NOT DISTINCT.
+     * @param allowNulls Matching null fields. Usually, NULL <> NULL in SQL. Except IS DISTINCT / IS NOT DISTINCT.
      * @return Rows comparator.
      */
-    Comparator<Row> comparator(List<RelFieldCollation> left, List<RelFieldCollation> right, boolean nullsEqual);
+    Comparator<Row> comparator(
+        List<RelFieldCollation> left,
+        List<RelFieldCollation> right,
+        ImmutableBitSet allowNulls
+    );
 
     /**
      * Creates a Filter predicate.

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/ExpressionFactoryImpl.java

@@ -62,6 +62,7 @@
 import org.apache.calcite.rex.RexUtil;
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.sql.validate.SqlConformance;
+import org.apache.calcite.util.ImmutableBitSet;
 import org.apache.ignite.internal.binary.BinaryUtils;
 import org.apache.ignite.internal.processors.query.calcite.exec.ExecutionContext;
 import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler;
@@ -197,7 +198,11 @@ else if (c2 != HIGHEST_VALUE)
     }
 
     /** {@inheritDoc} */
-    @Override public Comparator<Row> comparator(List<RelFieldCollation> left, List<RelFieldCollation> right, boolean nullsEqual) {
+    @Override public Comparator<Row> comparator(
+        List<RelFieldCollation> left,
+        List<RelFieldCollation> right,
+        ImmutableBitSet allowNulls
+    ) {
         if (F.isEmpty(left) || F.isEmpty(right) || left.size() != right.size())
             throw new IllegalArgumentException("Both inputs should be non-empty and have the same size: left="
                 + (left != null ? left.size() : "null") + ", right=" + (right != null ? right.size() : "null"));
@@ -213,9 +218,10 @@ else if (c2 != HIGHEST_VALUE)
 
         return new Comparator<Row>() {
             @Override public int compare(Row o1, Row o2) {
-                boolean hasNulls = false;
                 RowHandler<Row> hnd = ctx.rowHandler();
 
+                boolean hasNulls = false;
+
                 for (int i = 0; i < left.size(); i++) {
                     RelFieldCollation leftField = left.get(i);
                     RelFieldCollation rightField = right.get(i);
@@ -226,8 +232,9 @@ else if (c2 != HIGHEST_VALUE)
                     Object c1 = hnd.get(lIdx, o1);
                     Object c2 = hnd.get(rIdx, o2);
 
-                    if (c1 == null && c2 == null) {
-                        hasNulls = true;
+                    if (c1 == null && c2 == null && !hasNulls) {
+                        hasNulls = !allowNulls.get(i);
+
                         continue;
                     }
 
@@ -243,7 +250,7 @@ else if (c2 != HIGHEST_VALUE)
 
                 // If compared rows contain NULLs, they shouldn't be treated as equals, since NULL <> NULL in SQL.
                 // Except cases with IS DISTINCT / IS NOT DISTINCT.
-                return hasNulls && !nullsEqual ? 1 : 0;
+                return hasNulls ? 1 : 0;
             }
         };
     }

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/ModifyNode.java

@@ -42,6 +42,7 @@
 
 import static org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode.CONCURRENT_UPDATE;
 import static org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode.DUPLICATE_KEY;
+import static org.apache.ignite.internal.processors.query.QueryUtils.cacheForDML;
 
 /**
  *
@@ -218,7 +219,7 @@ private void invokeOutsideTransaction(
         GridCacheProxyImpl<Object, Object> cache
     ) throws IgniteCheckedException {
         Map<Object, EntryProcessor<Object, Object, Long>> map = invokeMap(tuples);
-        Map<Object, EntryProcessorResult<Long>> res = cache.invokeAll(map);
+        Map<Object, EntryProcessorResult<Long>> res = cacheForDML(cache).invokeAll(map);
 
         long updated = res.values().stream().mapToLong(EntryProcessorResult::get).sum();
 

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/message/InboxCloseMessage.java

@@ -17,23 +17,23 @@
 
 package org.apache.ignite.internal.processors.query.calcite.message;
 
-import java.nio.ByteBuffer;
 import java.util.UUID;
-
-import org.apache.ignite.plugin.extensions.communication.MessageReader;
-import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+import org.apache.ignite.internal.Order;
 
 /**
  *
  */
 public class InboxCloseMessage implements CalciteMessage {
     /** */
-    private UUID queryId;
+    @Order(value = 0, method = "queryId")
+    private UUID qryId;
 
     /** */
+    @Order(1)
     private long fragmentId;
 
     /** */
+    @Order(2)
     private long exchangeId;
 
     /** */
@@ -42,8 +42,8 @@ public InboxCloseMessage() {
     }
 
     /** */
-    public InboxCloseMessage(UUID queryId, long fragmentId, long exchangeId) {
-        this.queryId = queryId;
+    public InboxCloseMessage(UUID qryId, long fragmentId, long exchangeId) {
+        this.qryId = qryId;
         this.fragmentId = fragmentId;
         this.exchangeId = exchangeId;
     }
@@ -52,7 +52,14 @@ public InboxCloseMessage(UUID queryId, long fragmentId, long exchangeId) {
      * @return Query ID.
      */
     public UUID queryId() {
-        return queryId;
+        return qryId;
+    }
+
+    /**
+     * @param queryId New query ID.
+     */
+    public void queryId(UUID qryId) {
+        this.qryId = qryId;
     }
 
     /**
@@ -62,80 +69,25 @@ public long fragmentId() {
         return fragmentId;
     }
 
+    /**
+     * @param fragmentId New fragment ID.
+     */
+    public void fragmentId(long fragmentId) {
+        this.fragmentId = fragmentId;
+    }
+
     /**
      * @return Exchange ID.
      */
     public long exchangeId() {
         return exchangeId;
     }
 
-    /** {@inheritDoc} */
-    @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
-        writer.setBuffer(buf);
-
-        if (!writer.isHeaderWritten()) {
-            if (!writer.writeHeader(directType()))
-                return false;
-
-            writer.onHeaderWritten();
-        }
-
-        switch (writer.state()) {
-            case 0:
-                if (!writer.writeLong(exchangeId))
-                    return false;
-
-                writer.incrementState();
-
-            case 1:
-                if (!writer.writeLong(fragmentId))
-                    return false;
-
-                writer.incrementState();
-
-            case 2:
-                if (!writer.writeUuid(queryId))
-                    return false;
-
-                writer.incrementState();
-
-        }
-
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
-        reader.setBuffer(buf);
-
-        switch (reader.state()) {
-            case 0:
-                exchangeId = reader.readLong();
-
-                if (!reader.isLastRead())
-                    return false;
-
-                reader.incrementState();
-
-            case 1:
-                fragmentId = reader.readLong();
-
-                if (!reader.isLastRead())
-                    return false;
-
-                reader.incrementState();
-
-            case 2:
-                queryId = reader.readUuid();
-
-                if (!reader.isLastRead())
-                    return false;
-
-                reader.incrementState();
-
-        }
-
-        return true;
+    /**
+     * @param exchangeId New exchange ID.
+     */
+    public void exchangeId(long exchangeId) {
+        this.exchangeId = exchangeId;
     }
 
     /** {@inheritDoc} */