Copilot commented on code in PR #7463:
URL: https://github.com/apache/ignite-3/pull/7463#discussion_r2720803133
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/rule/CorrelatedNestedLoopJoinRule.java:
##########
@@ -84,10 +88,17 @@ public CorrelatedNestedLoopJoinRule(int batchSize) {
@Override
public RexNode visitInputRef(RexInputRef input) {
int field = input.getIndex();
+
if (field >= leftFieldCount) {
- return rexBuilder.makeInputRef(input.getType(),
input.getIndex() - leftFieldCount);
+ field = input.getIndex() - leftFieldCount;
+
+ corrRequiredColumnsBuilder.set(field);
+
+ return rexBuilder.makeInputRef(input.getType(), field);
}
+ corrRequiredColumnsBuilder.set(field);
+
Review Comment:
`corrRequiredColumnsBuilder` is meant to describe which *left-side* columns
must be propagated as correlation values (it is later used in
`CorrelatedNestedLoopJoinNode.prepareCorrelations()` to read fields from the
left row). In `field >= leftFieldCount` branch you currently set bits for
*right-side* input refs (after subtracting `leftFieldCount`), which will make
the join propagate incorrect columns (and can even cause out-of-bounds reads on
the left row). Only mark required correlation columns for the left-side refs
(the `else` branch).
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/SharedStateMessageConverter.java:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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.ignite.internal.sql.engine.exec;
+
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.calcite.avatica.util.ByteString;
+import org.apache.ignite.internal.network.NetworkMessage;
+import org.apache.ignite.internal.sql.engine.message.SharedStateMessage;
+import org.apache.ignite.internal.sql.engine.message.SqlQueryMessageGroup;
+import org.apache.ignite.internal.sql.engine.message.SqlQueryMessagesFactory;
+import org.apache.ignite.internal.sql.engine.message.field.SingleFieldMessage;
+import org.apache.ignite.internal.util.IgniteUtils;
+import org.jetbrains.annotations.Contract;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Converter between {@link SharedState} and {@link SharedStateMessage}.
+ */
+public class SharedStateMessageConverter {
+ /** Message factory. */
+ private static final SqlQueryMessagesFactory MESSAGE_FACTORY = new
SqlQueryMessagesFactory();
+
+ @Contract("null -> null; !null -> !null")
+ static @Nullable SharedStateMessage toMessage(@Nullable SharedState state)
{
+ if (state == null) {
+ return null;
+ }
+
+ Long2ObjectMap<Object> correlations = state.correlations();
+ Map<Long, NetworkMessage> result =
IgniteUtils.newHashMap(correlations.size());
+
+ for (Long2ObjectMap.Entry<Object> entry :
correlations.long2ObjectEntrySet()) {
+ SingleFieldMessage<?> msg = toSingleFieldMessage(entry.getValue());
+
+ result.put(entry.getLongKey(), msg);
+ }
+
+ return MESSAGE_FACTORY.sharedStateMessage()
+ .sharedState(result)
+ .build();
+ }
+
+ @Contract("null -> null; !null -> !null")
+ static @Nullable SharedState fromMessage(@Nullable SharedStateMessage
sharedStateMessage) {
+ if (sharedStateMessage == null) {
+ return null;
+ }
+
+ int size = sharedStateMessage.sharedState().size();
+ Long2ObjectMap<Object> correlations = new
Long2ObjectOpenHashMap<>(size);
+
+ for (Map.Entry<Long, NetworkMessage> e :
sharedStateMessage.sharedState().entrySet()) {
+ SingleFieldMessage<Object> msg = ((SingleFieldMessage<Object>)
e.getValue());
+
+ Object value = extractFieldValue(msg);
+
+ correlations.put(e.getKey().longValue(), value);
+ }
+
+ return new SharedState(correlations);
+ }
+
+ private static @Nullable Object
extractFieldValue(SingleFieldMessage<Object> msg) {
+ Object value = msg.field();
+
+ if (value == null) {
+ return null;
+ }
+
+ switch (msg.messageType()) {
+ case SqlQueryMessageGroup.BYTE_ARRAY_FIELD_MESSAGE:
+ return new ByteString((byte[]) value);
+
+ case SqlQueryMessageGroup.DECIMAL_FIELD_MESSAGE:
+ return decimalFromBytes((byte[]) value);
+
+ default:
+ return value;
+ }
+ }
+
+ private static BigDecimal decimalFromBytes(byte[] value) {
+ ByteBuffer buffer =
ByteBuffer.wrap(value).order(ByteOrder.LITTLE_ENDIAN);
+
+ short valScale = buffer.getShort();
+
+ BigInteger integer = new BigInteger(value, Short.BYTES, value.length -
Short.BYTES);
+
Review Comment:
`decimalFromBytes` uses `new BigInteger(value, Short.BYTES, value.length -
Short.BYTES)`, but `BigInteger` does not have a `(byte[], offset, length)`
constructor. This won’t compile and also doesn’t correctly slice out the
unscaled-value bytes. Extract the bytes after the scale (e.g., via a
slice/copy) and pass that to `new BigInteger(...)` (or use `ByteBuffer` to read
the remaining bytes) before constructing the `BigDecimal`.
```suggestion
byte[] unscaledBytes = new byte[value.length - Short.BYTES];
System.arraycopy(value, Short.BYTES, unscaledBytes, 0,
unscaledBytes.length);
BigInteger integer = new BigInteger(unscaledBytes);
```
##########
modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/SharedStateMessageConverter.java:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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.ignite.internal.sql.engine.exec;
+
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.calcite.avatica.util.ByteString;
+import org.apache.ignite.internal.network.NetworkMessage;
+import org.apache.ignite.internal.sql.engine.message.SharedStateMessage;
+import org.apache.ignite.internal.sql.engine.message.SqlQueryMessageGroup;
+import org.apache.ignite.internal.sql.engine.message.SqlQueryMessagesFactory;
+import org.apache.ignite.internal.sql.engine.message.field.SingleFieldMessage;
+import org.apache.ignite.internal.util.IgniteUtils;
+import org.jetbrains.annotations.Contract;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Converter between {@link SharedState} and {@link SharedStateMessage}.
+ */
+public class SharedStateMessageConverter {
+ /** Message factory. */
+ private static final SqlQueryMessagesFactory MESSAGE_FACTORY = new
SqlQueryMessagesFactory();
+
+ @Contract("null -> null; !null -> !null")
+ static @Nullable SharedStateMessage toMessage(@Nullable SharedState state)
{
+ if (state == null) {
+ return null;
+ }
+
+ Long2ObjectMap<Object> correlations = state.correlations();
+ Map<Long, NetworkMessage> result =
IgniteUtils.newHashMap(correlations.size());
+
+ for (Long2ObjectMap.Entry<Object> entry :
correlations.long2ObjectEntrySet()) {
+ SingleFieldMessage<?> msg = toSingleFieldMessage(entry.getValue());
+
+ result.put(entry.getLongKey(), msg);
+ }
+
+ return MESSAGE_FACTORY.sharedStateMessage()
+ .sharedState(result)
+ .build();
+ }
+
+ @Contract("null -> null; !null -> !null")
+ static @Nullable SharedState fromMessage(@Nullable SharedStateMessage
sharedStateMessage) {
+ if (sharedStateMessage == null) {
+ return null;
+ }
+
+ int size = sharedStateMessage.sharedState().size();
+ Long2ObjectMap<Object> correlations = new
Long2ObjectOpenHashMap<>(size);
+
+ for (Map.Entry<Long, NetworkMessage> e :
sharedStateMessage.sharedState().entrySet()) {
+ SingleFieldMessage<Object> msg = ((SingleFieldMessage<Object>)
e.getValue());
+
+ Object value = extractFieldValue(msg);
+
+ correlations.put(e.getKey().longValue(), value);
+ }
+
+ return new SharedState(correlations);
+ }
+
+ private static @Nullable Object
extractFieldValue(SingleFieldMessage<Object> msg) {
+ Object value = msg.field();
+
+ if (value == null) {
+ return null;
+ }
+
+ switch (msg.messageType()) {
+ case SqlQueryMessageGroup.BYTE_ARRAY_FIELD_MESSAGE:
+ return new ByteString((byte[]) value);
+
+ case SqlQueryMessageGroup.DECIMAL_FIELD_MESSAGE:
+ return decimalFromBytes((byte[]) value);
+
+ default:
+ return value;
+ }
+ }
+
+ private static BigDecimal decimalFromBytes(byte[] value) {
+ ByteBuffer buffer =
ByteBuffer.wrap(value).order(ByteOrder.LITTLE_ENDIAN);
+
+ short valScale = buffer.getShort();
+
+ BigInteger integer = new BigInteger(value, Short.BYTES, value.length -
Short.BYTES);
+
+ return new BigDecimal(integer, valScale);
+ }
+
+ private static byte[] decimalToBytes(BigDecimal value) {
+ byte[] unscaledBytes = value.unscaledValue().toByteArray();
+
+ ByteBuffer buffer = ByteBuffer.allocate(Short.BYTES +
unscaledBytes.length)
+ .order(ByteOrder.LITTLE_ENDIAN);
+
+ assert value.scale() <= Short.MAX_VALUE && value.scale() >=
Short.MIN_VALUE : "Scale out of range: " + value.scale();
+
+ buffer.putShort((short) value.scale());
Review Comment:
`decimalToBytes` relies on a Java `assert` to validate that
`BigDecimal.scale()` fits in a `short`. Asserts are typically disabled in
production, so an out-of-range scale would silently truncate when cast to
`short`, producing corrupted values on the wire. Please replace the assert with
an explicit runtime check that throws an exception (or otherwise handles
unsupported scales) before serializing.
```suggestion
int scale = value.scale();
if (scale > Short.MAX_VALUE || scale < Short.MIN_VALUE) {
throw new IllegalArgumentException("Scale out of range for
serialization: " + scale);
}
buffer.putShort((short) scale);
```
--
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]
