Copilot commented on code in PR #3459:
URL: https://github.com/apache/fluss/pull/3459#discussion_r3400620990
##########
fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java:
##########
@@ -355,17 +359,33 @@ public static List<TableChange>
toAddColumns(List<PbAddColumn> addColumns) {
return addColumns.stream()
.filter(Objects::nonNull)
.map(
- pbAddColumn ->
- TableChange.addColumn(
- pbAddColumn.getColumnName(),
- JsonSerdeUtils.readValue(
- pbAddColumn.getDataTypeJson(),
- DataTypeJsonSerde.INSTANCE),
- pbAddColumn.hasComment() ?
pbAddColumn.getComment() : null,
-
toColumnPosition(pbAddColumn.getColumnPositionType())))
+ pbAddColumn -> {
+ AggFunction aggFunction =
+ pbAddColumn.hasSerializedAggFunction()
+ ? deserializeAggFunction(
+
pbAddColumn.getSerializedAggFunction())
+ : null;
+ return TableChange.addColumn(
+ pbAddColumn.getColumnName(),
+ JsonSerdeUtils.readValue(
+ pbAddColumn.getDataTypeJson(),
+ DataTypeJsonSerde.INSTANCE),
+ pbAddColumn.hasComment() ?
pbAddColumn.getComment() : null,
+
toColumnPosition(pbAddColumn.getColumnPositionType()),
+ aggFunction);
+ })
.collect(Collectors.toList());
}
+ private static AggFunction deserializeAggFunction(byte[]
serializedAggFunction) {
+ try {
+ return InstantiationUtils.deserializeObject(
+ serializedAggFunction, AggFunction.class.getClassLoader());
+ } catch (IOException | ClassNotFoundException e) {
+ throw new FlussRuntimeException("Failed to deserialize aggregation
function.", e);
+ }
+ }
Review Comment:
Deserializing an arbitrary `byte[]` received over RPC into a Java object is
a high-risk pattern (gadget-based deserialization vulnerabilities) and is
sensitive to version/classloader compatibility. Prefer a safe, structured wire
format for aggregation functions (e.g., protobuf fields like function type +
parameters) or enforce strict class allowlisting / object input filtering
during deserialization (and reject any unexpected classes) so untrusted clients
cannot trigger unsafe deserialization.
##########
fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java:
##########
@@ -323,6 +323,9 @@ private static void checkArrowCompression(Configuration
tableConf) {
private static void checkMergeEngine(
Configuration tableConf, boolean hasPrimaryKey, Schema schema) {
MergeEngineType mergeEngine =
tableConf.get(ConfigOptions.TABLE_MERGE_ENGINE);
+ if (mergeEngine != MergeEngineType.AGGREGATION) {
+ validateNoAggregationFunctions(schema);
+ }
Review Comment:
This condition will also run when `mergeEngine` is `null` (since `null !=
MergeEngineType.AGGREGATION`), which can incorrectly reject schemas containing
aggregation functions even when the merge engine is unset/implicit. Consider
guarding the check with `mergeEngine != null && mergeEngine !=
MergeEngineType.AGGREGATION`, or otherwise aligning this validation with the
intended default merge-engine behavior.
##########
fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java:
##########
@@ -377,6 +380,20 @@ private static void checkMergeEngine(
}
}
+ /** Validates that the schema doesn't contain any aggregation functions. */
+ public static void validateNoAggregationFunctions(Schema schema) {
Review Comment:
Exposing this method as `public` broadens the API surface of
`TableDescriptorValidation` (even if it's in a server module) and makes it
harder to change later. If this is intended only for server-internal reuse
during schema alter, consider moving these validators into a dedicated internal
validator class (or annotating as internal if the project has such an
annotation) and/or minimizing visibility where possible to avoid treating it as
a supported external API.
--
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]
