You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "naivedogger (via GitHub)" <gi...@apache.org> on 2026/08/25 14:19:35 UTC
[PR] [FIP-49][gateway] Add schema-aware JSON and Arrow row codecs for FIP-49 data types [fluss]
naivedogger opened a new pull request, #4104:
URL: https://github.com/apache/fluss/pull/4104
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink], [rust], [python], [c++], [elixir]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
- **Generative AI disclosure:** Indicate whether generative AI tools were used in authoring this PR. If yes, specify the tool below.
- [ ] No generative AI tools used
- [ ] Yes (please specify the tool below)
**(The sections below can be removed for hotfixes or typos)**
-->
<!--
Generated-by: [Tool Name and Version] following [the guidelines](https://github.com/apache/fluss/blob/main/AGENTS.md)
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #4083
<!-- What is the purpose of the change -->
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FIP-49][gateway] Add schema-aware JSON and Arrow row codecs for FIP-49 data types [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on code in PR #4104:
URL: https://github.com/apache/fluss/pull/4104#discussion_r3870337078
##########
fluss-gateway/src/protocol/rest/codec/mod.rs:
##########
@@ -0,0 +1,35 @@
+// 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.
+
+//! Schema-aware conversion between FIP-49 JSON values and Fluss row representations.
+//!
+//! The write direction streams one raw JSON object through the table schema. Serde traverses
+//! containers once, numeric leaves retain their exact lexemes, and map visitors observe duplicate
+//! fields. String-keyed maps additionally accept JSON objects on input. The read direction renders
+//! Arrow values deterministically. BIGINT and DECIMAL use strings on output to avoid IEEE-754
+//! loss, binary values use base64, temporal values use ISO-8601, arrays and rows recurse, and maps
+//! use ordered `{"key", "value"}` entries on output so non-string keys and entry order survive.
+
+mod arrow_json;
+mod decoder;
+mod temporal;
+
+// Response assembly lands in a parallel task; keep the complete Arrow codec entry points available.
+#[allow(unused_imports)]
+pub(crate) use arrow_json::{record_batch_to_json_rows, value_to_json};
Review Comment:
LGTM overall. One scope note: since lookups are still listed as Future Work in #3957, the Arrow → JSON read direction has no consumer yet — I'd suggest dropping arrow_json.rs (plus the temporal.rs formatters and the arrow dependency that only serve it) and landing it with the lookup task instead.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FIP-49][gateway] Add schema-aware JSON row decoder [fluss]
Posted by "naivedogger (via GitHub)" <gi...@apache.org>.
naivedogger commented on code in PR #4104:
URL: https://github.com/apache/fluss/pull/4104#discussion_r3870634324
##########
fluss-gateway/src/protocol/rest/codec/mod.rs:
##########
@@ -0,0 +1,35 @@
+// 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.
+
+//! Schema-aware conversion between FIP-49 JSON values and Fluss row representations.
+//!
+//! The write direction streams one raw JSON object through the table schema. Serde traverses
+//! containers once, numeric leaves retain their exact lexemes, and map visitors observe duplicate
+//! fields. String-keyed maps additionally accept JSON objects on input. The read direction renders
+//! Arrow values deterministically. BIGINT and DECIMAL use strings on output to avoid IEEE-754
+//! loss, binary values use base64, temporal values use ISO-8601, arrays and rows recurse, and maps
+//! use ordered `{"key", "value"}` entries on output so non-string keys and entry order survive.
+
+mod arrow_json;
+mod decoder;
+mod temporal;
+
+// Response assembly lands in a parallel task; keep the complete Arrow codec entry points available.
+#[allow(unused_imports)]
+pub(crate) use arrow_json::{record_batch_to_json_rows, value_to_json};
Review Comment:
Thanks for the suggestion. This PR now focuses only on JSON-to-GenericRow decoding. We can add the Arrow-to-JSON codec later together with lookup support.
However, I noticed that #4083 currently includes the Arrow-to-JSON codec in its title and description. Since we’re moving that part to the future lookup task, should we update the scope of #4083 as well?
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FIP-49][gateway] Add schema-aware JSON and Arrow row codecs for FIP-49 data types [fluss]
Posted by "naivedogger (via GitHub)" <gi...@apache.org>.
naivedogger commented on PR #4104:
URL: https://github.com/apache/fluss/pull/4104#issuecomment-5435541697
Thanks @fresh-borzoni for the helpful review! I’ve addressed the comments, PTAL.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FIP-49][gateway] Add schema-aware JSON and Arrow row codecs for FIP-49 data types [fluss]
Posted by "naivedogger (via GitHub)" <gi...@apache.org>.
naivedogger commented on code in PR #4104:
URL: https://github.com/apache/fluss/pull/4104#discussion_r3869388365
##########
fluss-gateway/src/protocol/rest/codec/decoder.rs:
##########
@@ -0,0 +1,1927 @@
+// 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.
+
+//! Schema-aware streaming conversion from one raw FIP-49 JSON row to [`GenericRow`].
+//!
+//! Serde owns JSON syntax and container traversal. Schema-matching ARRAY, MAP, and ROW values
+//! recurse only along the validated [`RowType`], bounded by [`MAX_TYPE_NESTING`]. Scalar and
+//! ignored subtrees use serde_json's iterative [`RawValue`] and [`IgnoredAny`] scan path, so JSON
+//! depth does not grow this decoder's call stack. No container subtree is reparsed, keeping work
+//! linear in row bytes apart from local scalar decoding; the HTTP layer separately bounds body
+//! size. Recheck the iterative-scan assumption when upgrading serde_json.
+
+use crate::error::GatewayError;
+use crate::protocol::rest::codec::temporal::{parse_date, parse_time, parse_timestamp};
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64;
+use fluss::metadata::{DataType, RowType};
+use fluss::row::{
+ Date, Datum, Decimal, FlussArrayWriter, FlussMapWriter, GenericRow, Time, TimestampLtz,
+ TimestampNtz,
+};
+use serde::Deserialize;
+use serde::de::{self, DeserializeSeed, IgnoredAny, MapAccess, SeqAccess, Visitor};
+use serde_json::value::RawValue;
+use std::borrow::Cow;
+use std::collections::{HashMap, HashSet};
+use std::fmt;
+use std::sync::Arc;
+
+/// Bounds schema-recursive validation and decoding; it is not a JSON body-depth limit.
+const MAX_TYPE_NESTING: usize = 64;
+const NANOS_PER_MILLI: i64 = 1_000_000;
+const MILLIS_PER_SECOND: i64 = 1_000;
+const MILLIS_PER_DAY: i64 = 86_400_000;
+
+/// Whether a row supplies a complete mutation or only selected fields.
+#[derive(Debug, Clone, Copy)]
+#[allow(dead_code)]
+pub(crate) enum RowShape<'a> {
+ /// Every non-nullable column must be present. Missing nullable columns become null.
+ Complete,
+ /// Only the named columns must be present and non-null. Other missing columns become null.
+ Sparse(&'a [String]),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum RowDecodeErrorKind {
+ SchemaMismatch,
+ InvalidValue,
+}
+
+/// One row decode failure plus whether refreshed metadata could plausibly resolve it.
+#[derive(Debug)]
+pub(crate) struct RowDecodeError {
+ kind: RowDecodeErrorKind,
+ error: GatewayError,
+}
+
+#[allow(dead_code)]
+impl RowDecodeError {
+ fn schema_mismatch(error: GatewayError) -> Self {
+ Self {
+ kind: RowDecodeErrorKind::SchemaMismatch,
+ error,
+ }
+ }
+
+ fn invalid(error: GatewayError) -> Self {
+ Self {
+ kind: RowDecodeErrorKind::InvalidValue,
+ error,
+ }
+ }
+
+ /// Returns true only for failures that a newer table schema may resolve.
+ pub(crate) fn is_schema_mismatch(&self) -> bool {
+ self.kind == RowDecodeErrorKind::SchemaMismatch
+ }
+
+ /// Returns the client-safe error message.
+ pub(crate) fn message(&self) -> &str {
+ self.error.message()
+ }
+
+ /// Converts this failure into the standard Gateway error envelope input.
+ pub(crate) fn into_gateway_error(self) -> GatewayError {
+ self.error
+ }
+}
+
+impl From<GatewayError> for RowDecodeError {
+ fn from(error: GatewayError) -> Self {
+ Self::invalid(error)
+ }
+}
+
+struct SchemaDecoderInner {
+ row_type: RowType,
+ column_indexes: HashMap<String, usize>,
+}
+
+/// Immutable reusable decoder compiled from one Fluss row type.
+#[derive(Clone)]
+#[allow(dead_code)]
+pub(crate) struct SchemaDecoder {
+ inner: Arc<SchemaDecoderInner>,
+}
+
+impl fmt::Debug for SchemaDecoder {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter
+ .debug_struct("SchemaDecoder")
+ .field("row_type", &self.inner.row_type)
+ .finish()
+ }
+}
+
+#[allow(dead_code)]
+impl SchemaDecoder {
+ /// Builds a decoder for a complete table row type.
+ pub(crate) fn new(row_type: RowType) -> Result<Self, GatewayError> {
+ validate_row_type(&row_type, 0)?;
+ if row_type.fields().is_empty() {
+ return Err(GatewayError::internal(
+ "cannot build a row decoder for an empty table schema",
+ ));
+ }
+ let column_indexes = row_type
+ .fields()
+ .iter()
+ .enumerate()
+ .map(|(index, field)| (field.name.clone(), index))
+ .collect();
+ Ok(Self {
+ inner: Arc::new(SchemaDecoderInner {
+ row_type,
+ column_indexes,
+ }),
+ })
+ }
+
+ /// Returns the exact row type this decoder was built from.
+ pub(crate) fn row_type(&self) -> &RowType {
+ &self.inner.row_type
+ }
+
+ /// Decodes one raw JSON object into complete schema order.
+ pub(crate) fn decode_row(
+ &self,
+ label: &str,
+ row_json: &[u8],
+ shape: RowShape<'_>,
+ ) -> Result<GenericRow<'static>, RowDecodeError> {
+ let required = self.required_columns(label, shape)?;
+ let seed = RootRowSeed {
+ label,
+ row_type: &self.inner.row_type,
+ column_indexes: &self.inner.column_indexes,
+ required: &required,
+ };
+ let mut deserializer = serde_json::Deserializer::from_slice(row_json);
+ let decoded = seed
+ .deserialize(&mut deserializer)
+ .map_err(|error| invalid_json_error(label, error))?;
+ deserializer
+ .end()
+ .map_err(|error| invalid_json_error(label, error))?;
+ decoded
+ }
+
+ fn required_columns(
+ &self,
+ label: &str,
+ shape: RowShape<'_>,
+ ) -> Result<Vec<bool>, RowDecodeError> {
+ match shape {
+ RowShape::Complete => Ok(self
+ .inner
+ .row_type
+ .fields()
+ .iter()
+ .map(|field| !field.data_type.is_nullable())
+ .collect()),
+ RowShape::Sparse(required_columns) => {
+ let mut required = vec![false; self.inner.row_type.fields().len()];
+ for column in required_columns {
+ let Some(index) = self.inner.column_indexes.get(column) else {
+ return Err(RowDecodeError::schema_mismatch(
+ GatewayError::invalid_argument(format!(
+ "{label}: required column `{column}` is not part of the table schema"
+ )),
+ ));
+ };
+ required[*index] = true;
+ }
+ Ok(required)
+ }
+ }
+ }
+}
+
+fn invalid_json_error(label: &str, error: serde_json::Error) -> RowDecodeError {
+ RowDecodeError::invalid(GatewayError::invalid_argument(format!(
+ "{label}: invalid JSON row: {error}"
+ )))
+}
+
+fn validate_row_type(row_type: &RowType, depth: usize) -> Result<(), GatewayError> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::internal(format!(
+ "row type nesting exceeds {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ let mut names = HashSet::with_capacity(row_type.fields().len());
+ for field in row_type.fields() {
+ if field.name.is_empty() {
+ return Err(GatewayError::internal(
+ "row type contains an empty field name",
+ ));
+ }
+ if !names.insert(field.name.as_str()) {
+ return Err(GatewayError::internal(format!(
+ "row type contains duplicate field `{}`",
+ field.name
+ )));
+ }
+ validate_data_type(&field.data_type, depth + 1)?;
+ }
+ Ok(())
+}
+
+fn validate_data_type(data_type: &DataType, depth: usize) -> Result<(), GatewayError> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::internal(format!(
+ "row type nesting exceeds {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ match data_type {
+ DataType::Char(data_type) if data_type.length() == 0 => {
+ Err(GatewayError::internal("CHAR length must be at least one"))
+ }
+ DataType::Binary(data_type) if data_type.length() == 0 => {
+ Err(GatewayError::internal("BINARY length must be at least one"))
+ }
+ DataType::Array(data_type) => validate_data_type(data_type.get_element_type(), depth + 1),
+ DataType::Map(data_type) => {
+ if data_type.key_type().is_nullable() {
+ return Err(GatewayError::internal("MAP key type must not be nullable"));
+ }
+ if is_complex(data_type.key_type()) {
+ return Err(GatewayError::invalid_argument(
+ "FIP-49 JSON does not support ARRAY, MAP, or ROW values as MAP keys",
+ ));
+ }
+ validate_data_type(data_type.key_type(), depth + 1)?;
+ validate_data_type(data_type.value_type(), depth + 1)
+ }
+ DataType::Row(row_type) => validate_row_type(row_type, depth + 1),
+ _ => Ok(()),
+ }
+}
+
+fn is_complex(data_type: &DataType) -> bool {
+ matches!(
+ data_type,
+ DataType::Array(_) | DataType::Map(_) | DataType::Row(_)
+ )
+}
+
+#[derive(Clone)]
+struct ValuePath<'a> {
+ label: &'a str,
+ path: String,
+}
+
+impl<'a> ValuePath<'a> {
+ fn column(label: &'a str, column: &str) -> Self {
+ Self {
+ label,
+ path: column.to_string(),
+ }
+ }
+
+ fn element(&self, index: usize) -> Self {
+ self.nested(format!("{}[{index}]", self.path))
+ }
+
+ fn nested_field(&self, name: &str) -> Self {
+ self.nested(format!("{}.{name}", self.path))
+ }
+
+ fn map_part(&self, index: usize, part: &str) -> Self {
+ self.nested(format!("{}[{index}].{part}", self.path))
+ }
+
+ fn nested(&self, path: String) -> Self {
+ Self {
+ label: self.label,
+ path,
+ }
+ }
+
+ fn invalid(&self, reason: impl fmt::Display) -> RowDecodeError {
+ RowDecodeError::invalid(GatewayError::invalid_argument(format!(
+ "{}: column `{}` {reason}",
+ self.label, self.path
+ )))
+ }
+
+ fn schema_mismatch(&self, reason: impl fmt::Display) -> RowDecodeError {
+ RowDecodeError::schema_mismatch(GatewayError::invalid_argument(format!(
+ "{}: column `{}` {reason}",
+ self.label, self.path
+ )))
+ }
+
+ fn type_error(&self, expected: &str, actual: &str) -> RowDecodeError {
+ self.invalid(format!("expects {expected}, got {actual}"))
+ }
+}
+
+type DecodeResult<T> = Result<T, RowDecodeError>;
+
+enum ScalarValue<'a> {
+ Null,
+ Boolean(bool),
+ ExactNumber(&'a str),
+ String(String),
+ Array,
+ Object,
+}
+
+impl ScalarValue<'_> {
+ fn kind(&self) -> &'static str {
+ match self {
+ Self::Null => "null",
+ Self::Boolean(_) => "a boolean",
+ Self::ExactNumber(_) => "a number",
+ Self::String(_) => "a string",
+ Self::Array => "an array",
+ Self::Object => "an object",
+ }
+ }
+}
+
+fn parse_scalar_value(raw: &RawValue) -> Result<ScalarValue<'_>, serde_json::Error> {
+ let text = raw.get().trim();
+ match text.as_bytes().first().copied() {
+ Some(b'n') => Ok(ScalarValue::Null),
+ Some(b't') => Ok(ScalarValue::Boolean(true)),
+ Some(b'f') => Ok(ScalarValue::Boolean(false)),
+ Some(b'"') => serde_json::from_str(text).map(ScalarValue::String),
+ Some(b'-' | b'0'..=b'9') => Ok(ScalarValue::ExactNumber(text)),
+ Some(b'[') => Ok(ScalarValue::Array),
+ Some(b'{') => Ok(ScalarValue::Object),
+ _ => unreachable!("RawValue always contains one complete JSON value"),
+ }
+}
+
+struct RootRowSeed<'a> {
+ label: &'a str,
+ row_type: &'a RowType,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+}
+
+impl<'de> DeserializeSeed<'de> for RootRowSeed<'_> {
+ type Value = DecodeResult<GenericRow<'static>>;
+
+ fn deserialize<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
+ where
+ D: de::Deserializer<'de>,
+ {
+ deserializer.deserialize_any(RootRowVisitor {
+ label: self.label,
+ row_type: self.row_type,
+ column_indexes: self.column_indexes,
+ required: self.required,
+ })
+ }
+}
+
+struct RootRowVisitor<'a> {
+ label: &'a str,
+ row_type: &'a RowType,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+}
+
+impl RootRowVisitor<'_> {
+ fn type_error(&self, actual: &str) -> DecodeResult<GenericRow<'static>> {
+ Err(RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{}: row must be a JSON object, got {actual}", self.label),
+ )))
+ }
+}
+
+impl<'de> Visitor<'de> for RootRowVisitor<'_> {
+ type Value = DecodeResult<GenericRow<'static>>;
+
+ fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str("a JSON row object")
+ }
+
+ fn visit_unit<E>(self) -> Result<Self::Value, E> {
+ Ok(self.type_error("null"))
+ }
+
+ fn visit_bool<E>(self, _value: bool) -> Result<Self::Value, E> {
+ Ok(self.type_error("a boolean"))
+ }
+
+ fn visit_i64<E>(self, _value: i64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_u64<E>(self, _value: u64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_f64<E>(self, _value: f64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_str<E>(self, _value: &str) -> Result<Self::Value, E> {
+ Ok(self.type_error("a string"))
+ }
+
+ fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
+ where
+ A: SeqAccess<'de>,
+ {
+ drain_sequence(&mut sequence)?;
+ Ok(self.type_error("an array"))
+ }
+
+ fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>
+ where
+ A: MapAccess<'de>,
+ {
+ decode_row_map(
+ map,
+ self.row_type,
+ RowContext::Root {
+ label: self.label,
+ column_indexes: self.column_indexes,
+ required: self.required,
+ },
+ )
+ }
+}
+
+enum RowContext<'a> {
+ Root {
+ label: &'a str,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+ },
+ Nested {
+ path: ValuePath<'a>,
+ },
+}
+
+impl<'a> RowContext<'a> {
+ fn field_index(&self, row_type: &RowType, name: &str) -> Option<usize> {
+ match self {
+ Self::Root { column_indexes, .. } => column_indexes.get(name).copied(),
+ Self::Nested { .. } => row_type.get_field_index(name),
+ }
+ }
+
+ fn field_path(&self, name: &str) -> ValuePath<'a> {
+ match self {
+ Self::Root { label, .. } => ValuePath::column(label, name),
+ Self::Nested { path } => path.nested_field(name),
+ }
+ }
+
+ fn required(&self, index: usize, data_type: &DataType) -> bool {
+ match self {
+ Self::Root { required, .. } => required[index],
+ Self::Nested { .. } => !data_type.is_nullable(),
+ }
+ }
+
+ fn duplicate(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{label}: duplicate column `{name}`"),
+ )),
+ Self::Nested { path } => path.invalid(format!("has a duplicate field `{name}`")),
+ }
+ }
+
+ fn unknown(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::schema_mismatch(
+ GatewayError::invalid_argument(format!("{label}: unknown column `{name}`")),
+ ),
+ Self::Nested { path } => path.schema_mismatch(format!("has an unknown field `{name}`")),
+ }
+ }
+
+ fn missing(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => {
+ RowDecodeError::schema_mismatch(GatewayError::invalid_argument(format!(
+ "{label}: column `{name}` is required and was not provided"
+ )))
+ }
+ Self::Nested { path } => path
+ .nested_field(name)
+ .schema_mismatch("is required and was not provided"),
+ }
+ }
+
+ fn required_null(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{label}: column `{name}` is required and must not be null"),
+ )),
+ Self::Nested { path } => path.nested_field(name).invalid("must not be null"),
+ }
+ }
+}
+
+fn decode_row_map<'de, A>(
+ mut map: A,
+ row_type: &RowType,
+ context: RowContext<'_>,
+) -> Result<DecodeResult<GenericRow<'static>>, A::Error>
+where
+ A: MapAccess<'de>,
+{
+ let mut seen = HashSet::with_capacity(map.size_hint().unwrap_or(0));
+ let mut duplicate = None;
+ let mut unknown = None;
+ let mut provided = std::iter::repeat_with(|| None)
+ .take(row_type.fields().len())
+ .collect::<Vec<Option<DecodeResult<Datum<'static>>>>>();
+
+ while let Some(name) = map.next_key::<String>()? {
+ if !seen.insert(name.clone()) {
+ map.next_value::<IgnoredAny>()?;
+ duplicate.get_or_insert(name);
+ continue;
+ }
+ let Some(index) = context.field_index(row_type, &name) else {
+ map.next_value::<IgnoredAny>()?;
+ unknown.get_or_insert(name);
+ continue;
+ };
+ let field = &row_type.fields()[index];
+ let value = map.next_value_seed(ValueSeed {
+ path: context.field_path(&field.name),
+ data_type: &field.data_type,
+ })?;
+ provided[index] = Some(value);
+ }
+
+ if let Some(name) = duplicate {
+ return Ok(Err(context.duplicate(&name)));
+ }
+ if let Some(name) = unknown {
+ return Ok(Err(context.unknown(&name)));
+ }
+
+ let mut row = GenericRow::new(row_type.fields().len());
+ for (index, field) in row_type.fields().iter().enumerate() {
+ let datum = match provided[index].take() {
+ Some(Ok(Datum::Null)) if context.required(index, &field.data_type) => {
+ return Ok(Err(context.required_null(&field.name)));
+ }
+ Some(Ok(datum)) => datum,
+ Some(Err(error)) => return Ok(Err(error)),
+ None if context.required(index, &field.data_type) => {
+ return Ok(Err(context.missing(&field.name)));
+ }
+ None => Datum::Null,
+ };
+ row.set_field(index, datum);
+ }
+ Ok(Ok(row))
+}
+
+struct ValueSeed<'a> {
+ path: ValuePath<'a>,
+ data_type: &'a DataType,
+}
+
+impl<'de> DeserializeSeed<'de> for ValueSeed<'_> {
+ type Value = DecodeResult<Datum<'static>>;
+
+ fn deserialize<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
+ where
+ D: de::Deserializer<'de>,
+ {
+ if is_complex(self.data_type) {
+ deserializer.deserialize_any(ComplexValueVisitor {
+ path: self.path,
+ data_type: self.data_type,
+ })
+ } else {
+ let raw = <&RawValue>::deserialize(deserializer)?;
+ let value = parse_scalar_value(raw).map_err(de::Error::custom)?;
+ Ok(decode_scalar(&self.path, self.data_type, &value))
+ }
+ }
+}
+
+struct ComplexValueVisitor<'a> {
+ path: ValuePath<'a>,
+ data_type: &'a DataType,
+}
+
+impl ComplexValueVisitor<'_> {
+ fn type_error(&self, actual: &str) -> DecodeResult<Datum<'static>> {
+ Err(self
+ .path
+ .type_error(complex_expected(self.data_type), actual))
+ }
+
+ fn null(&self) -> DecodeResult<Datum<'static>> {
+ if self.data_type.is_nullable() {
+ Ok(Datum::Null)
+ } else {
+ Err(self.path.invalid("must not be null"))
+ }
+ }
+}
+
+impl<'de> Visitor<'de> for ComplexValueVisitor<'_> {
+ type Value = DecodeResult<Datum<'static>>;
+
+ fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str(complex_expected(self.data_type))
+ }
+
+ fn visit_unit<E>(self) -> Result<Self::Value, E> {
+ Ok(self.null())
+ }
+
+ fn visit_bool<E>(self, _value: bool) -> Result<Self::Value, E> {
+ Ok(self.type_error("a boolean"))
+ }
+
+ fn visit_i64<E>(self, _value: i64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_u64<E>(self, _value: u64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_f64<E>(self, _value: f64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_str<E>(self, _value: &str) -> Result<Self::Value, E> {
+ Ok(self.type_error("a string"))
+ }
+
+ fn visit_seq<A>(self, sequence: A) -> Result<Self::Value, A::Error>
+ where
+ A: SeqAccess<'de>,
+ {
+ match self.data_type {
+ DataType::Array(data_type) => {
+ decode_array_sequence(sequence, &self.path, data_type.get_element_type())
+ }
+ DataType::Map(data_type) => decode_map_sequence(
+ sequence,
+ &self.path,
+ data_type.key_type(),
+ data_type.value_type(),
+ ),
+ DataType::Row(_) => {
+ let mut sequence = sequence;
+ drain_sequence(&mut sequence)?;
+ Ok(self.type_error("an array"))
+ }
+ _ => unreachable!("complex visitor receives only ARRAY, MAP, or ROW"),
+ }
+ }
+
+ fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>
+ where
+ A: MapAccess<'de>,
+ {
+ match self.data_type {
+ DataType::Row(row_type) => {
+ let row = decode_row_map(map, row_type, RowContext::Nested { path: self.path })?;
+ Ok(row.map(|row| Datum::Row(Box::new(row))))
+ }
+ DataType::Array(_) | DataType::Map(_) => {
+ let mut map = map;
+ drain_map(&mut map)?;
+ Ok(self.type_error("an object"))
+ }
+ _ => unreachable!("complex visitor receives only ARRAY, MAP, or ROW"),
+ }
+ }
+}
+
+fn complex_expected(data_type: &DataType) -> &'static str {
+ match data_type {
+ DataType::Array(_) => "ARRAY (a JSON array)",
+ DataType::Map(_) => "MAP (an array of {key, value} objects)",
Review Comment:
Agreed. MAP<STRING, V> should accept both forms, while non-string keys still require the entry-array form. I’ve added object-form input support and kept the entry-array form as the canonical output.
##########
fluss-gateway/src/protocol/rest/codec/temporal.rs:
##########
@@ -0,0 +1,252 @@
+// 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.
+
+//! Shared temporal parsing and formatting for both codec directions.
+
+const DAYS_PER_GREGORIAN_CYCLE: i64 = 146_097;
+const DAYS_FROM_CIVIL_EPOCH_TO_UNIX_EPOCH: i64 = 719_468;
+
+pub(super) struct ParsedTimestamp {
+ pub(super) days: i64,
+ pub(super) seconds_of_day: i64,
+ pub(super) frac_nanos: i64,
+ pub(super) offset_seconds: Option<i32>,
+}
+
+pub(super) fn parse_date(text: &str) -> Option<i64> {
+ let (negative, unsigned) = match text.strip_prefix('-') {
+ Some(rest) => (true, rest),
+ None => (false, text.strip_prefix('+').unwrap_or(text)),
+ };
+ let (year, month_and_day) = unsigned.split_once('-')?;
+ let (month, day) = month_and_day.split_once('-')?;
+ if year.len() < 4
+ || month.len() != 2
+ || day.len() != 2
+ || month_and_day.matches('-').count() != 1
+ {
+ return None;
+ }
+ let unsigned_year: i64 = digits_only(year)?.parse().ok()?;
+ let year = if negative {
+ unsigned_year.checked_neg()?
+ } else {
+ unsigned_year
+ };
+ let month: u32 = digits_only(month)?.parse().ok()?;
+ let day: u32 = digits_only(day)?.parse().ok()?;
+ if !(1..=12).contains(&month) || !(1..=days_in_month(year, month)).contains(&day) {
+ return None;
+ }
+ days_from_civil(year, month, day)
+}
+
+pub(super) fn parse_time(text: &str) -> Option<(i64, i64)> {
+ let (clock, fraction) = match text.split_once('.') {
+ Some((clock, fraction)) => (clock, Some(fraction)),
+ None => (text, None),
+ };
+ let bytes = clock.as_bytes();
+ let (hour, minute, second) = match bytes.len() {
+ 5 if bytes[2] == b':' => (clock.get(0..2)?, clock.get(3..5)?, "0"),
+ 8 if bytes[2] == b':' && bytes[5] == b':' => {
+ (clock.get(0..2)?, clock.get(3..5)?, clock.get(6..8)?)
+ }
+ _ => return None,
+ };
+ let hour: i64 = digits_only(hour)?.parse().ok()?;
+ let minute: i64 = digits_only(minute)?.parse().ok()?;
+ let second: i64 = digits_only(second)?.parse().ok()?;
+ if hour > 23 || minute > 59 || second > 59 {
+ return None;
+ }
+ let frac_nanos = match fraction {
+ None => 0,
+ Some(fraction) => {
+ if fraction.is_empty() || fraction.len() > 9 {
+ return None;
+ }
+ let digits: i64 = digits_only(fraction)?.parse().ok()?;
+ digits * 10_i64.pow(9 - fraction.len() as u32)
+ }
+ };
+ Some((hour * 3_600 + minute * 60 + second, frac_nanos))
+}
+
+pub(super) fn parse_timestamp(text: &str) -> Option<ParsedTimestamp> {
+ let separator = text.find(['T', ' '])?;
Review Comment:
Good catch. Since the FIP specifies ISO-8601 without defining a space-separated extension, I agree that we should only accept T.
##########
fluss-gateway/src/protocol/rest/codec/decoder.rs:
##########
@@ -0,0 +1,1927 @@
+// 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.
+
+//! Schema-aware streaming conversion from one raw FIP-49 JSON row to [`GenericRow`].
+//!
+//! Serde owns JSON syntax and container traversal. Schema-matching ARRAY, MAP, and ROW values
+//! recurse only along the validated [`RowType`], bounded by [`MAX_TYPE_NESTING`]. Scalar and
+//! ignored subtrees use serde_json's iterative [`RawValue`] and [`IgnoredAny`] scan path, so JSON
+//! depth does not grow this decoder's call stack. No container subtree is reparsed, keeping work
+//! linear in row bytes apart from local scalar decoding; the HTTP layer separately bounds body
+//! size. Recheck the iterative-scan assumption when upgrading serde_json.
+
+use crate::error::GatewayError;
+use crate::protocol::rest::codec::temporal::{parse_date, parse_time, parse_timestamp};
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64;
+use fluss::metadata::{DataType, RowType};
+use fluss::row::{
+ Date, Datum, Decimal, FlussArrayWriter, FlussMapWriter, GenericRow, Time, TimestampLtz,
+ TimestampNtz,
+};
+use serde::Deserialize;
+use serde::de::{self, DeserializeSeed, IgnoredAny, MapAccess, SeqAccess, Visitor};
+use serde_json::value::RawValue;
+use std::borrow::Cow;
+use std::collections::{HashMap, HashSet};
+use std::fmt;
+use std::sync::Arc;
+
+/// Bounds schema-recursive validation and decoding; it is not a JSON body-depth limit.
+const MAX_TYPE_NESTING: usize = 64;
+const NANOS_PER_MILLI: i64 = 1_000_000;
+const MILLIS_PER_SECOND: i64 = 1_000;
+const MILLIS_PER_DAY: i64 = 86_400_000;
+
+/// Whether a row supplies a complete mutation or only selected fields.
+#[derive(Debug, Clone, Copy)]
+#[allow(dead_code)]
+pub(crate) enum RowShape<'a> {
+ /// Every non-nullable column must be present. Missing nullable columns become null.
+ Complete,
+ /// Only the named columns must be present and non-null. Other missing columns become null.
+ Sparse(&'a [String]),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum RowDecodeErrorKind {
+ SchemaMismatch,
+ InvalidValue,
+}
+
+/// One row decode failure plus whether refreshed metadata could plausibly resolve it.
+#[derive(Debug)]
+pub(crate) struct RowDecodeError {
+ kind: RowDecodeErrorKind,
+ error: GatewayError,
+}
+
+#[allow(dead_code)]
+impl RowDecodeError {
+ fn schema_mismatch(error: GatewayError) -> Self {
+ Self {
+ kind: RowDecodeErrorKind::SchemaMismatch,
+ error,
+ }
+ }
+
+ fn invalid(error: GatewayError) -> Self {
+ Self {
+ kind: RowDecodeErrorKind::InvalidValue,
+ error,
+ }
+ }
+
+ /// Returns true only for failures that a newer table schema may resolve.
+ pub(crate) fn is_schema_mismatch(&self) -> bool {
+ self.kind == RowDecodeErrorKind::SchemaMismatch
+ }
+
+ /// Returns the client-safe error message.
+ pub(crate) fn message(&self) -> &str {
+ self.error.message()
+ }
+
+ /// Converts this failure into the standard Gateway error envelope input.
+ pub(crate) fn into_gateway_error(self) -> GatewayError {
+ self.error
+ }
+}
+
+impl From<GatewayError> for RowDecodeError {
+ fn from(error: GatewayError) -> Self {
+ Self::invalid(error)
+ }
+}
+
+struct SchemaDecoderInner {
+ row_type: RowType,
+ column_indexes: HashMap<String, usize>,
+}
+
+/// Immutable reusable decoder compiled from one Fluss row type.
+#[derive(Clone)]
+#[allow(dead_code)]
+pub(crate) struct SchemaDecoder {
+ inner: Arc<SchemaDecoderInner>,
+}
+
+impl fmt::Debug for SchemaDecoder {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter
+ .debug_struct("SchemaDecoder")
+ .field("row_type", &self.inner.row_type)
+ .finish()
+ }
+}
+
+#[allow(dead_code)]
+impl SchemaDecoder {
+ /// Builds a decoder for a complete table row type.
+ pub(crate) fn new(row_type: RowType) -> Result<Self, GatewayError> {
+ validate_row_type(&row_type, 0)?;
+ if row_type.fields().is_empty() {
+ return Err(GatewayError::internal(
+ "cannot build a row decoder for an empty table schema",
+ ));
+ }
+ let column_indexes = row_type
+ .fields()
+ .iter()
+ .enumerate()
+ .map(|(index, field)| (field.name.clone(), index))
+ .collect();
+ Ok(Self {
+ inner: Arc::new(SchemaDecoderInner {
+ row_type,
+ column_indexes,
+ }),
+ })
+ }
+
+ /// Returns the exact row type this decoder was built from.
+ pub(crate) fn row_type(&self) -> &RowType {
+ &self.inner.row_type
+ }
+
+ /// Decodes one raw JSON object into complete schema order.
+ pub(crate) fn decode_row(
+ &self,
+ label: &str,
+ row_json: &[u8],
+ shape: RowShape<'_>,
+ ) -> Result<GenericRow<'static>, RowDecodeError> {
+ let required = self.required_columns(label, shape)?;
+ let seed = RootRowSeed {
+ label,
+ row_type: &self.inner.row_type,
+ column_indexes: &self.inner.column_indexes,
+ required: &required,
+ };
+ let mut deserializer = serde_json::Deserializer::from_slice(row_json);
+ let decoded = seed
+ .deserialize(&mut deserializer)
+ .map_err(|error| invalid_json_error(label, error))?;
+ deserializer
+ .end()
+ .map_err(|error| invalid_json_error(label, error))?;
+ decoded
+ }
+
+ fn required_columns(
+ &self,
+ label: &str,
+ shape: RowShape<'_>,
+ ) -> Result<Vec<bool>, RowDecodeError> {
+ match shape {
+ RowShape::Complete => Ok(self
+ .inner
+ .row_type
+ .fields()
+ .iter()
+ .map(|field| !field.data_type.is_nullable())
+ .collect()),
+ RowShape::Sparse(required_columns) => {
+ let mut required = vec![false; self.inner.row_type.fields().len()];
+ for column in required_columns {
+ let Some(index) = self.inner.column_indexes.get(column) else {
+ return Err(RowDecodeError::schema_mismatch(
+ GatewayError::invalid_argument(format!(
+ "{label}: required column `{column}` is not part of the table schema"
+ )),
+ ));
+ };
+ required[*index] = true;
+ }
+ Ok(required)
+ }
+ }
+ }
+}
+
+fn invalid_json_error(label: &str, error: serde_json::Error) -> RowDecodeError {
+ RowDecodeError::invalid(GatewayError::invalid_argument(format!(
+ "{label}: invalid JSON row: {error}"
+ )))
+}
+
+fn validate_row_type(row_type: &RowType, depth: usize) -> Result<(), GatewayError> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::internal(format!(
+ "row type nesting exceeds {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ let mut names = HashSet::with_capacity(row_type.fields().len());
+ for field in row_type.fields() {
+ if field.name.is_empty() {
+ return Err(GatewayError::internal(
+ "row type contains an empty field name",
+ ));
+ }
+ if !names.insert(field.name.as_str()) {
+ return Err(GatewayError::internal(format!(
+ "row type contains duplicate field `{}`",
+ field.name
+ )));
+ }
+ validate_data_type(&field.data_type, depth + 1)?;
+ }
+ Ok(())
+}
+
+fn validate_data_type(data_type: &DataType, depth: usize) -> Result<(), GatewayError> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::internal(format!(
+ "row type nesting exceeds {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ match data_type {
+ DataType::Char(data_type) if data_type.length() == 0 => {
+ Err(GatewayError::internal("CHAR length must be at least one"))
+ }
+ DataType::Binary(data_type) if data_type.length() == 0 => {
+ Err(GatewayError::internal("BINARY length must be at least one"))
+ }
+ DataType::Array(data_type) => validate_data_type(data_type.get_element_type(), depth + 1),
+ DataType::Map(data_type) => {
+ if data_type.key_type().is_nullable() {
+ return Err(GatewayError::internal("MAP key type must not be nullable"));
+ }
+ if is_complex(data_type.key_type()) {
+ return Err(GatewayError::invalid_argument(
+ "FIP-49 JSON does not support ARRAY, MAP, or ROW values as MAP keys",
+ ));
+ }
+ validate_data_type(data_type.key_type(), depth + 1)?;
+ validate_data_type(data_type.value_type(), depth + 1)
+ }
+ DataType::Row(row_type) => validate_row_type(row_type, depth + 1),
+ _ => Ok(()),
+ }
+}
+
+fn is_complex(data_type: &DataType) -> bool {
+ matches!(
+ data_type,
+ DataType::Array(_) | DataType::Map(_) | DataType::Row(_)
+ )
+}
+
+#[derive(Clone)]
+struct ValuePath<'a> {
+ label: &'a str,
+ path: String,
+}
+
+impl<'a> ValuePath<'a> {
+ fn column(label: &'a str, column: &str) -> Self {
+ Self {
+ label,
+ path: column.to_string(),
+ }
+ }
+
+ fn element(&self, index: usize) -> Self {
+ self.nested(format!("{}[{index}]", self.path))
+ }
+
+ fn nested_field(&self, name: &str) -> Self {
+ self.nested(format!("{}.{name}", self.path))
+ }
+
+ fn map_part(&self, index: usize, part: &str) -> Self {
+ self.nested(format!("{}[{index}].{part}", self.path))
+ }
+
+ fn nested(&self, path: String) -> Self {
+ Self {
+ label: self.label,
+ path,
+ }
+ }
+
+ fn invalid(&self, reason: impl fmt::Display) -> RowDecodeError {
+ RowDecodeError::invalid(GatewayError::invalid_argument(format!(
+ "{}: column `{}` {reason}",
+ self.label, self.path
+ )))
+ }
+
+ fn schema_mismatch(&self, reason: impl fmt::Display) -> RowDecodeError {
+ RowDecodeError::schema_mismatch(GatewayError::invalid_argument(format!(
+ "{}: column `{}` {reason}",
+ self.label, self.path
+ )))
+ }
+
+ fn type_error(&self, expected: &str, actual: &str) -> RowDecodeError {
+ self.invalid(format!("expects {expected}, got {actual}"))
+ }
+}
+
+type DecodeResult<T> = Result<T, RowDecodeError>;
+
+enum ScalarValue<'a> {
+ Null,
+ Boolean(bool),
+ ExactNumber(&'a str),
+ String(String),
+ Array,
+ Object,
+}
+
+impl ScalarValue<'_> {
+ fn kind(&self) -> &'static str {
+ match self {
+ Self::Null => "null",
+ Self::Boolean(_) => "a boolean",
+ Self::ExactNumber(_) => "a number",
+ Self::String(_) => "a string",
+ Self::Array => "an array",
+ Self::Object => "an object",
+ }
+ }
+}
+
+fn parse_scalar_value(raw: &RawValue) -> Result<ScalarValue<'_>, serde_json::Error> {
+ let text = raw.get().trim();
+ match text.as_bytes().first().copied() {
+ Some(b'n') => Ok(ScalarValue::Null),
+ Some(b't') => Ok(ScalarValue::Boolean(true)),
+ Some(b'f') => Ok(ScalarValue::Boolean(false)),
+ Some(b'"') => serde_json::from_str(text).map(ScalarValue::String),
+ Some(b'-' | b'0'..=b'9') => Ok(ScalarValue::ExactNumber(text)),
+ Some(b'[') => Ok(ScalarValue::Array),
+ Some(b'{') => Ok(ScalarValue::Object),
+ _ => unreachable!("RawValue always contains one complete JSON value"),
+ }
+}
+
+struct RootRowSeed<'a> {
+ label: &'a str,
+ row_type: &'a RowType,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+}
+
+impl<'de> DeserializeSeed<'de> for RootRowSeed<'_> {
+ type Value = DecodeResult<GenericRow<'static>>;
+
+ fn deserialize<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
+ where
+ D: de::Deserializer<'de>,
+ {
+ deserializer.deserialize_any(RootRowVisitor {
+ label: self.label,
+ row_type: self.row_type,
+ column_indexes: self.column_indexes,
+ required: self.required,
+ })
+ }
+}
+
+struct RootRowVisitor<'a> {
+ label: &'a str,
+ row_type: &'a RowType,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+}
+
+impl RootRowVisitor<'_> {
+ fn type_error(&self, actual: &str) -> DecodeResult<GenericRow<'static>> {
+ Err(RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{}: row must be a JSON object, got {actual}", self.label),
+ )))
+ }
+}
+
+impl<'de> Visitor<'de> for RootRowVisitor<'_> {
+ type Value = DecodeResult<GenericRow<'static>>;
+
+ fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str("a JSON row object")
+ }
+
+ fn visit_unit<E>(self) -> Result<Self::Value, E> {
+ Ok(self.type_error("null"))
+ }
+
+ fn visit_bool<E>(self, _value: bool) -> Result<Self::Value, E> {
+ Ok(self.type_error("a boolean"))
+ }
+
+ fn visit_i64<E>(self, _value: i64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_u64<E>(self, _value: u64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_f64<E>(self, _value: f64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_str<E>(self, _value: &str) -> Result<Self::Value, E> {
+ Ok(self.type_error("a string"))
+ }
+
+ fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
+ where
+ A: SeqAccess<'de>,
+ {
+ drain_sequence(&mut sequence)?;
+ Ok(self.type_error("an array"))
+ }
+
+ fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>
+ where
+ A: MapAccess<'de>,
+ {
+ decode_row_map(
+ map,
+ self.row_type,
+ RowContext::Root {
+ label: self.label,
+ column_indexes: self.column_indexes,
+ required: self.required,
+ },
+ )
+ }
+}
+
+enum RowContext<'a> {
+ Root {
+ label: &'a str,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+ },
+ Nested {
+ path: ValuePath<'a>,
+ },
+}
+
+impl<'a> RowContext<'a> {
+ fn field_index(&self, row_type: &RowType, name: &str) -> Option<usize> {
+ match self {
+ Self::Root { column_indexes, .. } => column_indexes.get(name).copied(),
+ Self::Nested { .. } => row_type.get_field_index(name),
+ }
+ }
+
+ fn field_path(&self, name: &str) -> ValuePath<'a> {
+ match self {
+ Self::Root { label, .. } => ValuePath::column(label, name),
+ Self::Nested { path } => path.nested_field(name),
+ }
+ }
+
+ fn required(&self, index: usize, data_type: &DataType) -> bool {
+ match self {
+ Self::Root { required, .. } => required[index],
+ Self::Nested { .. } => !data_type.is_nullable(),
+ }
+ }
+
+ fn duplicate(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{label}: duplicate column `{name}`"),
+ )),
+ Self::Nested { path } => path.invalid(format!("has a duplicate field `{name}`")),
+ }
+ }
+
+ fn unknown(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::schema_mismatch(
+ GatewayError::invalid_argument(format!("{label}: unknown column `{name}`")),
+ ),
+ Self::Nested { path } => path.schema_mismatch(format!("has an unknown field `{name}`")),
+ }
+ }
+
+ fn missing(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => {
+ RowDecodeError::schema_mismatch(GatewayError::invalid_argument(format!(
+ "{label}: column `{name}` is required and was not provided"
+ )))
+ }
+ Self::Nested { path } => path
+ .nested_field(name)
+ .schema_mismatch("is required and was not provided"),
+ }
+ }
+
+ fn required_null(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{label}: column `{name}` is required and must not be null"),
+ )),
+ Self::Nested { path } => path.nested_field(name).invalid("must not be null"),
+ }
+ }
+}
+
+fn decode_row_map<'de, A>(
+ mut map: A,
+ row_type: &RowType,
+ context: RowContext<'_>,
+) -> Result<DecodeResult<GenericRow<'static>>, A::Error>
+where
+ A: MapAccess<'de>,
+{
+ let mut seen = HashSet::with_capacity(map.size_hint().unwrap_or(0));
+ let mut duplicate = None;
+ let mut unknown = None;
+ let mut provided = std::iter::repeat_with(|| None)
+ .take(row_type.fields().len())
+ .collect::<Vec<Option<DecodeResult<Datum<'static>>>>>();
+
+ while let Some(name) = map.next_key::<String>()? {
+ if !seen.insert(name.clone()) {
+ map.next_value::<IgnoredAny>()?;
+ duplicate.get_or_insert(name);
+ continue;
+ }
+ let Some(index) = context.field_index(row_type, &name) else {
+ map.next_value::<IgnoredAny>()?;
+ unknown.get_or_insert(name);
+ continue;
+ };
+ let field = &row_type.fields()[index];
+ let value = map.next_value_seed(ValueSeed {
+ path: context.field_path(&field.name),
+ data_type: &field.data_type,
+ })?;
+ provided[index] = Some(value);
+ }
+
+ if let Some(name) = duplicate {
+ return Ok(Err(context.duplicate(&name)));
+ }
+ if let Some(name) = unknown {
+ return Ok(Err(context.unknown(&name)));
+ }
+
+ let mut row = GenericRow::new(row_type.fields().len());
+ for (index, field) in row_type.fields().iter().enumerate() {
+ let datum = match provided[index].take() {
+ Some(Ok(Datum::Null)) if context.required(index, &field.data_type) => {
+ return Ok(Err(context.required_null(&field.name)));
+ }
+ Some(Ok(datum)) => datum,
+ Some(Err(error)) => return Ok(Err(error)),
+ None if context.required(index, &field.data_type) => {
+ return Ok(Err(context.missing(&field.name)));
+ }
+ None => Datum::Null,
+ };
+ row.set_field(index, datum);
+ }
+ Ok(Ok(row))
+}
+
+struct ValueSeed<'a> {
+ path: ValuePath<'a>,
+ data_type: &'a DataType,
+}
+
+impl<'de> DeserializeSeed<'de> for ValueSeed<'_> {
+ type Value = DecodeResult<Datum<'static>>;
+
+ fn deserialize<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
+ where
+ D: de::Deserializer<'de>,
+ {
+ if is_complex(self.data_type) {
+ deserializer.deserialize_any(ComplexValueVisitor {
+ path: self.path,
+ data_type: self.data_type,
+ })
+ } else {
+ let raw = <&RawValue>::deserialize(deserializer)?;
+ let value = parse_scalar_value(raw).map_err(de::Error::custom)?;
+ Ok(decode_scalar(&self.path, self.data_type, &value))
+ }
+ }
+}
+
+struct ComplexValueVisitor<'a> {
+ path: ValuePath<'a>,
+ data_type: &'a DataType,
+}
+
+impl ComplexValueVisitor<'_> {
+ fn type_error(&self, actual: &str) -> DecodeResult<Datum<'static>> {
+ Err(self
+ .path
+ .type_error(complex_expected(self.data_type), actual))
+ }
+
+ fn null(&self) -> DecodeResult<Datum<'static>> {
+ if self.data_type.is_nullable() {
+ Ok(Datum::Null)
+ } else {
+ Err(self.path.invalid("must not be null"))
+ }
+ }
+}
+
+impl<'de> Visitor<'de> for ComplexValueVisitor<'_> {
+ type Value = DecodeResult<Datum<'static>>;
+
+ fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str(complex_expected(self.data_type))
+ }
+
+ fn visit_unit<E>(self) -> Result<Self::Value, E> {
+ Ok(self.null())
+ }
+
+ fn visit_bool<E>(self, _value: bool) -> Result<Self::Value, E> {
+ Ok(self.type_error("a boolean"))
+ }
+
+ fn visit_i64<E>(self, _value: i64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_u64<E>(self, _value: u64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_f64<E>(self, _value: f64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_str<E>(self, _value: &str) -> Result<Self::Value, E> {
+ Ok(self.type_error("a string"))
+ }
+
+ fn visit_seq<A>(self, sequence: A) -> Result<Self::Value, A::Error>
+ where
+ A: SeqAccess<'de>,
+ {
+ match self.data_type {
+ DataType::Array(data_type) => {
+ decode_array_sequence(sequence, &self.path, data_type.get_element_type())
+ }
+ DataType::Map(data_type) => decode_map_sequence(
+ sequence,
+ &self.path,
+ data_type.key_type(),
+ data_type.value_type(),
+ ),
+ DataType::Row(_) => {
+ let mut sequence = sequence;
+ drain_sequence(&mut sequence)?;
+ Ok(self.type_error("an array"))
+ }
+ _ => unreachable!("complex visitor receives only ARRAY, MAP, or ROW"),
+ }
+ }
+
+ fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>
+ where
+ A: MapAccess<'de>,
+ {
+ match self.data_type {
+ DataType::Row(row_type) => {
+ let row = decode_row_map(map, row_type, RowContext::Nested { path: self.path })?;
+ Ok(row.map(|row| Datum::Row(Box::new(row))))
+ }
+ DataType::Array(_) | DataType::Map(_) => {
+ let mut map = map;
+ drain_map(&mut map)?;
+ Ok(self.type_error("an object"))
+ }
+ _ => unreachable!("complex visitor receives only ARRAY, MAP, or ROW"),
+ }
+ }
+}
+
+fn complex_expected(data_type: &DataType) -> &'static str {
+ match data_type {
+ DataType::Array(_) => "ARRAY (a JSON array)",
+ DataType::Map(_) => "MAP (an array of {key, value} objects)",
+ DataType::Row(_) => "ROW (a JSON object of its declared fields)",
+ _ => unreachable!("only complex data types have a container expectation"),
+ }
+}
+
+fn drain_sequence<'de, A>(sequence: &mut A) -> Result<(), A::Error>
+where
+ A: SeqAccess<'de>,
+{
+ while sequence.next_element::<IgnoredAny>()?.is_some() {}
+ Ok(())
+}
+
+fn drain_map<'de, A>(map: &mut A) -> Result<(), A::Error>
+where
+ A: MapAccess<'de>,
+{
+ while map.next_entry::<IgnoredAny, IgnoredAny>()?.is_some() {}
+ Ok(())
+}
+
+fn decode_scalar(
+ path: &ValuePath<'_>,
+ data_type: &DataType,
+ value: &ScalarValue<'_>,
+) -> DecodeResult<Datum<'static>> {
+ if matches!(value, ScalarValue::Null) {
+ return if data_type.is_nullable() {
+ Ok(Datum::Null)
+ } else {
+ Err(path.invalid("must not be null"))
+ };
+ }
+ match data_type {
+ DataType::Boolean(_) => match value {
+ ScalarValue::Boolean(parsed) => Ok(Datum::Bool(*parsed)),
+ _ => Err(path.type_error("BOOLEAN (a JSON boolean)", value.kind())),
+ },
+ DataType::TinyInt(_) => {
+ decode_integer(path, value, "TINYINT", i8::MIN as i64, i8::MAX as i64)
+ .map(|parsed| Datum::Int8(parsed as i8))
+ }
+ DataType::SmallInt(_) => {
+ decode_integer(path, value, "SMALLINT", i16::MIN as i64, i16::MAX as i64)
+ .map(|parsed| Datum::Int16(parsed as i16))
+ }
+ DataType::Int(_) => decode_integer(path, value, "INT", i32::MIN as i64, i32::MAX as i64)
+ .map(|parsed| Datum::Int32(parsed as i32)),
+ DataType::BigInt(_) => decode_bigint(path, value).map(Datum::Int64),
+ DataType::Float(_) => decode_float32(path, value),
+ DataType::Double(_) => decode_float(path, value, "DOUBLE").map(Datum::from),
+ DataType::Char(_) | DataType::String(_) => match value {
Review Comment:
Agreed. CHAR(n) now rejects values longer than n Unicode code points, while shorter values remain valid. I also added coverage for shorter, multibyte, boundary, and oversized values.
##########
fluss-gateway/src/protocol/rest/codec/decoder.rs:
##########
@@ -0,0 +1,1927 @@
+// 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.
+
+//! Schema-aware streaming conversion from one raw FIP-49 JSON row to [`GenericRow`].
+//!
+//! Serde owns JSON syntax and container traversal. Schema-matching ARRAY, MAP, and ROW values
+//! recurse only along the validated [`RowType`], bounded by [`MAX_TYPE_NESTING`]. Scalar and
+//! ignored subtrees use serde_json's iterative [`RawValue`] and [`IgnoredAny`] scan path, so JSON
+//! depth does not grow this decoder's call stack. No container subtree is reparsed, keeping work
+//! linear in row bytes apart from local scalar decoding; the HTTP layer separately bounds body
+//! size. Recheck the iterative-scan assumption when upgrading serde_json.
+
+use crate::error::GatewayError;
+use crate::protocol::rest::codec::temporal::{parse_date, parse_time, parse_timestamp};
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64;
+use fluss::metadata::{DataType, RowType};
+use fluss::row::{
+ Date, Datum, Decimal, FlussArrayWriter, FlussMapWriter, GenericRow, Time, TimestampLtz,
+ TimestampNtz,
+};
+use serde::Deserialize;
+use serde::de::{self, DeserializeSeed, IgnoredAny, MapAccess, SeqAccess, Visitor};
+use serde_json::value::RawValue;
+use std::borrow::Cow;
+use std::collections::{HashMap, HashSet};
+use std::fmt;
+use std::sync::Arc;
+
+/// Bounds schema-recursive validation and decoding; it is not a JSON body-depth limit.
+const MAX_TYPE_NESTING: usize = 64;
+const NANOS_PER_MILLI: i64 = 1_000_000;
+const MILLIS_PER_SECOND: i64 = 1_000;
+const MILLIS_PER_DAY: i64 = 86_400_000;
+
+/// Whether a row supplies a complete mutation or only selected fields.
+#[derive(Debug, Clone, Copy)]
+#[allow(dead_code)]
+pub(crate) enum RowShape<'a> {
+ /// Every non-nullable column must be present. Missing nullable columns become null.
+ Complete,
+ /// Only the named columns must be present and non-null. Other missing columns become null.
+ Sparse(&'a [String]),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum RowDecodeErrorKind {
+ SchemaMismatch,
+ InvalidValue,
+}
+
+/// One row decode failure plus whether refreshed metadata could plausibly resolve it.
+#[derive(Debug)]
+pub(crate) struct RowDecodeError {
+ kind: RowDecodeErrorKind,
+ error: GatewayError,
+}
+
+#[allow(dead_code)]
+impl RowDecodeError {
+ fn schema_mismatch(error: GatewayError) -> Self {
+ Self {
+ kind: RowDecodeErrorKind::SchemaMismatch,
+ error,
+ }
+ }
+
+ fn invalid(error: GatewayError) -> Self {
+ Self {
+ kind: RowDecodeErrorKind::InvalidValue,
+ error,
+ }
+ }
+
+ /// Returns true only for failures that a newer table schema may resolve.
+ pub(crate) fn is_schema_mismatch(&self) -> bool {
+ self.kind == RowDecodeErrorKind::SchemaMismatch
+ }
+
+ /// Returns the client-safe error message.
+ pub(crate) fn message(&self) -> &str {
+ self.error.message()
+ }
+
+ /// Converts this failure into the standard Gateway error envelope input.
+ pub(crate) fn into_gateway_error(self) -> GatewayError {
+ self.error
+ }
+}
+
+impl From<GatewayError> for RowDecodeError {
+ fn from(error: GatewayError) -> Self {
+ Self::invalid(error)
+ }
+}
+
+struct SchemaDecoderInner {
+ row_type: RowType,
+ column_indexes: HashMap<String, usize>,
+}
+
+/// Immutable reusable decoder compiled from one Fluss row type.
+#[derive(Clone)]
+#[allow(dead_code)]
+pub(crate) struct SchemaDecoder {
+ inner: Arc<SchemaDecoderInner>,
+}
+
+impl fmt::Debug for SchemaDecoder {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter
+ .debug_struct("SchemaDecoder")
+ .field("row_type", &self.inner.row_type)
+ .finish()
+ }
+}
+
+#[allow(dead_code)]
+impl SchemaDecoder {
+ /// Builds a decoder for a complete table row type.
+ pub(crate) fn new(row_type: RowType) -> Result<Self, GatewayError> {
+ validate_row_type(&row_type, 0)?;
+ if row_type.fields().is_empty() {
+ return Err(GatewayError::internal(
+ "cannot build a row decoder for an empty table schema",
+ ));
+ }
+ let column_indexes = row_type
+ .fields()
+ .iter()
+ .enumerate()
+ .map(|(index, field)| (field.name.clone(), index))
+ .collect();
+ Ok(Self {
+ inner: Arc::new(SchemaDecoderInner {
+ row_type,
+ column_indexes,
+ }),
+ })
+ }
+
+ /// Returns the exact row type this decoder was built from.
+ pub(crate) fn row_type(&self) -> &RowType {
+ &self.inner.row_type
+ }
+
+ /// Decodes one raw JSON object into complete schema order.
+ pub(crate) fn decode_row(
+ &self,
+ label: &str,
+ row_json: &[u8],
+ shape: RowShape<'_>,
+ ) -> Result<GenericRow<'static>, RowDecodeError> {
+ let required = self.required_columns(label, shape)?;
+ let seed = RootRowSeed {
+ label,
+ row_type: &self.inner.row_type,
+ column_indexes: &self.inner.column_indexes,
+ required: &required,
+ };
+ let mut deserializer = serde_json::Deserializer::from_slice(row_json);
+ let decoded = seed
+ .deserialize(&mut deserializer)
+ .map_err(|error| invalid_json_error(label, error))?;
+ deserializer
+ .end()
+ .map_err(|error| invalid_json_error(label, error))?;
+ decoded
+ }
+
+ fn required_columns(
+ &self,
+ label: &str,
+ shape: RowShape<'_>,
+ ) -> Result<Vec<bool>, RowDecodeError> {
+ match shape {
+ RowShape::Complete => Ok(self
+ .inner
+ .row_type
+ .fields()
+ .iter()
+ .map(|field| !field.data_type.is_nullable())
+ .collect()),
+ RowShape::Sparse(required_columns) => {
+ let mut required = vec![false; self.inner.row_type.fields().len()];
+ for column in required_columns {
+ let Some(index) = self.inner.column_indexes.get(column) else {
+ return Err(RowDecodeError::schema_mismatch(
+ GatewayError::invalid_argument(format!(
+ "{label}: required column `{column}` is not part of the table schema"
+ )),
+ ));
+ };
+ required[*index] = true;
+ }
+ Ok(required)
+ }
+ }
+ }
+}
+
+fn invalid_json_error(label: &str, error: serde_json::Error) -> RowDecodeError {
+ RowDecodeError::invalid(GatewayError::invalid_argument(format!(
+ "{label}: invalid JSON row: {error}"
+ )))
+}
+
+fn validate_row_type(row_type: &RowType, depth: usize) -> Result<(), GatewayError> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::internal(format!(
+ "row type nesting exceeds {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ let mut names = HashSet::with_capacity(row_type.fields().len());
+ for field in row_type.fields() {
+ if field.name.is_empty() {
+ return Err(GatewayError::internal(
+ "row type contains an empty field name",
+ ));
+ }
+ if !names.insert(field.name.as_str()) {
+ return Err(GatewayError::internal(format!(
+ "row type contains duplicate field `{}`",
+ field.name
+ )));
+ }
+ validate_data_type(&field.data_type, depth + 1)?;
+ }
+ Ok(())
+}
+
+fn validate_data_type(data_type: &DataType, depth: usize) -> Result<(), GatewayError> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::internal(format!(
+ "row type nesting exceeds {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ match data_type {
+ DataType::Char(data_type) if data_type.length() == 0 => {
+ Err(GatewayError::internal("CHAR length must be at least one"))
+ }
+ DataType::Binary(data_type) if data_type.length() == 0 => {
+ Err(GatewayError::internal("BINARY length must be at least one"))
+ }
+ DataType::Array(data_type) => validate_data_type(data_type.get_element_type(), depth + 1),
+ DataType::Map(data_type) => {
+ if data_type.key_type().is_nullable() {
+ return Err(GatewayError::internal("MAP key type must not be nullable"));
+ }
+ if is_complex(data_type.key_type()) {
+ return Err(GatewayError::invalid_argument(
+ "FIP-49 JSON does not support ARRAY, MAP, or ROW values as MAP keys",
+ ));
+ }
+ validate_data_type(data_type.key_type(), depth + 1)?;
+ validate_data_type(data_type.value_type(), depth + 1)
+ }
+ DataType::Row(row_type) => validate_row_type(row_type, depth + 1),
+ _ => Ok(()),
+ }
+}
+
+fn is_complex(data_type: &DataType) -> bool {
+ matches!(
+ data_type,
+ DataType::Array(_) | DataType::Map(_) | DataType::Row(_)
+ )
+}
+
+#[derive(Clone)]
+struct ValuePath<'a> {
+ label: &'a str,
+ path: String,
+}
+
+impl<'a> ValuePath<'a> {
+ fn column(label: &'a str, column: &str) -> Self {
+ Self {
+ label,
+ path: column.to_string(),
+ }
+ }
+
+ fn element(&self, index: usize) -> Self {
+ self.nested(format!("{}[{index}]", self.path))
+ }
+
+ fn nested_field(&self, name: &str) -> Self {
+ self.nested(format!("{}.{name}", self.path))
+ }
+
+ fn map_part(&self, index: usize, part: &str) -> Self {
+ self.nested(format!("{}[{index}].{part}", self.path))
+ }
+
+ fn nested(&self, path: String) -> Self {
+ Self {
+ label: self.label,
+ path,
+ }
+ }
+
+ fn invalid(&self, reason: impl fmt::Display) -> RowDecodeError {
+ RowDecodeError::invalid(GatewayError::invalid_argument(format!(
+ "{}: column `{}` {reason}",
+ self.label, self.path
+ )))
+ }
+
+ fn schema_mismatch(&self, reason: impl fmt::Display) -> RowDecodeError {
+ RowDecodeError::schema_mismatch(GatewayError::invalid_argument(format!(
+ "{}: column `{}` {reason}",
+ self.label, self.path
+ )))
+ }
+
+ fn type_error(&self, expected: &str, actual: &str) -> RowDecodeError {
+ self.invalid(format!("expects {expected}, got {actual}"))
+ }
+}
+
+type DecodeResult<T> = Result<T, RowDecodeError>;
+
+enum ScalarValue<'a> {
+ Null,
+ Boolean(bool),
+ ExactNumber(&'a str),
+ String(String),
+ Array,
+ Object,
+}
+
+impl ScalarValue<'_> {
+ fn kind(&self) -> &'static str {
+ match self {
+ Self::Null => "null",
+ Self::Boolean(_) => "a boolean",
+ Self::ExactNumber(_) => "a number",
+ Self::String(_) => "a string",
+ Self::Array => "an array",
+ Self::Object => "an object",
+ }
+ }
+}
+
+fn parse_scalar_value(raw: &RawValue) -> Result<ScalarValue<'_>, serde_json::Error> {
+ let text = raw.get().trim();
+ match text.as_bytes().first().copied() {
+ Some(b'n') => Ok(ScalarValue::Null),
+ Some(b't') => Ok(ScalarValue::Boolean(true)),
+ Some(b'f') => Ok(ScalarValue::Boolean(false)),
+ Some(b'"') => serde_json::from_str(text).map(ScalarValue::String),
+ Some(b'-' | b'0'..=b'9') => Ok(ScalarValue::ExactNumber(text)),
+ Some(b'[') => Ok(ScalarValue::Array),
+ Some(b'{') => Ok(ScalarValue::Object),
+ _ => unreachable!("RawValue always contains one complete JSON value"),
+ }
+}
+
+struct RootRowSeed<'a> {
+ label: &'a str,
+ row_type: &'a RowType,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+}
+
+impl<'de> DeserializeSeed<'de> for RootRowSeed<'_> {
+ type Value = DecodeResult<GenericRow<'static>>;
+
+ fn deserialize<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
+ where
+ D: de::Deserializer<'de>,
+ {
+ deserializer.deserialize_any(RootRowVisitor {
+ label: self.label,
+ row_type: self.row_type,
+ column_indexes: self.column_indexes,
+ required: self.required,
+ })
+ }
+}
+
+struct RootRowVisitor<'a> {
+ label: &'a str,
+ row_type: &'a RowType,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+}
+
+impl RootRowVisitor<'_> {
+ fn type_error(&self, actual: &str) -> DecodeResult<GenericRow<'static>> {
+ Err(RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{}: row must be a JSON object, got {actual}", self.label),
+ )))
+ }
+}
+
+impl<'de> Visitor<'de> for RootRowVisitor<'_> {
+ type Value = DecodeResult<GenericRow<'static>>;
+
+ fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str("a JSON row object")
+ }
+
+ fn visit_unit<E>(self) -> Result<Self::Value, E> {
+ Ok(self.type_error("null"))
+ }
+
+ fn visit_bool<E>(self, _value: bool) -> Result<Self::Value, E> {
+ Ok(self.type_error("a boolean"))
+ }
+
+ fn visit_i64<E>(self, _value: i64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_u64<E>(self, _value: u64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_f64<E>(self, _value: f64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_str<E>(self, _value: &str) -> Result<Self::Value, E> {
+ Ok(self.type_error("a string"))
+ }
+
+ fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
+ where
+ A: SeqAccess<'de>,
+ {
+ drain_sequence(&mut sequence)?;
+ Ok(self.type_error("an array"))
+ }
+
+ fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>
+ where
+ A: MapAccess<'de>,
+ {
+ decode_row_map(
+ map,
+ self.row_type,
+ RowContext::Root {
+ label: self.label,
+ column_indexes: self.column_indexes,
+ required: self.required,
+ },
+ )
+ }
+}
+
+enum RowContext<'a> {
+ Root {
+ label: &'a str,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+ },
+ Nested {
+ path: ValuePath<'a>,
+ },
+}
+
+impl<'a> RowContext<'a> {
+ fn field_index(&self, row_type: &RowType, name: &str) -> Option<usize> {
+ match self {
+ Self::Root { column_indexes, .. } => column_indexes.get(name).copied(),
+ Self::Nested { .. } => row_type.get_field_index(name),
+ }
+ }
+
+ fn field_path(&self, name: &str) -> ValuePath<'a> {
+ match self {
+ Self::Root { label, .. } => ValuePath::column(label, name),
+ Self::Nested { path } => path.nested_field(name),
+ }
+ }
+
+ fn required(&self, index: usize, data_type: &DataType) -> bool {
+ match self {
+ Self::Root { required, .. } => required[index],
+ Self::Nested { .. } => !data_type.is_nullable(),
+ }
+ }
+
+ fn duplicate(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{label}: duplicate column `{name}`"),
+ )),
+ Self::Nested { path } => path.invalid(format!("has a duplicate field `{name}`")),
+ }
+ }
+
+ fn unknown(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::schema_mismatch(
+ GatewayError::invalid_argument(format!("{label}: unknown column `{name}`")),
+ ),
+ Self::Nested { path } => path.schema_mismatch(format!("has an unknown field `{name}`")),
+ }
+ }
+
+ fn missing(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => {
+ RowDecodeError::schema_mismatch(GatewayError::invalid_argument(format!(
+ "{label}: column `{name}` is required and was not provided"
+ )))
+ }
+ Self::Nested { path } => path
+ .nested_field(name)
+ .schema_mismatch("is required and was not provided"),
+ }
+ }
+
+ fn required_null(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{label}: column `{name}` is required and must not be null"),
+ )),
+ Self::Nested { path } => path.nested_field(name).invalid("must not be null"),
+ }
+ }
+}
+
+fn decode_row_map<'de, A>(
+ mut map: A,
+ row_type: &RowType,
+ context: RowContext<'_>,
+) -> Result<DecodeResult<GenericRow<'static>>, A::Error>
+where
+ A: MapAccess<'de>,
+{
+ let mut seen = HashSet::with_capacity(map.size_hint().unwrap_or(0));
+ let mut duplicate = None;
+ let mut unknown = None;
+ let mut provided = std::iter::repeat_with(|| None)
+ .take(row_type.fields().len())
+ .collect::<Vec<Option<DecodeResult<Datum<'static>>>>>();
+
+ while let Some(name) = map.next_key::<String>()? {
+ if !seen.insert(name.clone()) {
+ map.next_value::<IgnoredAny>()?;
+ duplicate.get_or_insert(name);
+ continue;
+ }
+ let Some(index) = context.field_index(row_type, &name) else {
+ map.next_value::<IgnoredAny>()?;
+ unknown.get_or_insert(name);
+ continue;
+ };
+ let field = &row_type.fields()[index];
+ let value = map.next_value_seed(ValueSeed {
+ path: context.field_path(&field.name),
+ data_type: &field.data_type,
+ })?;
+ provided[index] = Some(value);
+ }
+
+ if let Some(name) = duplicate {
+ return Ok(Err(context.duplicate(&name)));
+ }
+ if let Some(name) = unknown {
+ return Ok(Err(context.unknown(&name)));
+ }
+
+ let mut row = GenericRow::new(row_type.fields().len());
+ for (index, field) in row_type.fields().iter().enumerate() {
+ let datum = match provided[index].take() {
+ Some(Ok(Datum::Null)) if context.required(index, &field.data_type) => {
+ return Ok(Err(context.required_null(&field.name)));
+ }
+ Some(Ok(datum)) => datum,
+ Some(Err(error)) => return Ok(Err(error)),
+ None if context.required(index, &field.data_type) => {
+ return Ok(Err(context.missing(&field.name)));
+ }
+ None => Datum::Null,
+ };
+ row.set_field(index, datum);
+ }
+ Ok(Ok(row))
+}
+
+struct ValueSeed<'a> {
+ path: ValuePath<'a>,
+ data_type: &'a DataType,
+}
+
+impl<'de> DeserializeSeed<'de> for ValueSeed<'_> {
+ type Value = DecodeResult<Datum<'static>>;
+
+ fn deserialize<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
+ where
+ D: de::Deserializer<'de>,
+ {
+ if is_complex(self.data_type) {
+ deserializer.deserialize_any(ComplexValueVisitor {
+ path: self.path,
+ data_type: self.data_type,
+ })
+ } else {
+ let raw = <&RawValue>::deserialize(deserializer)?;
+ let value = parse_scalar_value(raw).map_err(de::Error::custom)?;
+ Ok(decode_scalar(&self.path, self.data_type, &value))
+ }
+ }
+}
+
+struct ComplexValueVisitor<'a> {
+ path: ValuePath<'a>,
+ data_type: &'a DataType,
+}
+
+impl ComplexValueVisitor<'_> {
+ fn type_error(&self, actual: &str) -> DecodeResult<Datum<'static>> {
+ Err(self
+ .path
+ .type_error(complex_expected(self.data_type), actual))
+ }
+
+ fn null(&self) -> DecodeResult<Datum<'static>> {
+ if self.data_type.is_nullable() {
+ Ok(Datum::Null)
+ } else {
+ Err(self.path.invalid("must not be null"))
+ }
+ }
+}
+
+impl<'de> Visitor<'de> for ComplexValueVisitor<'_> {
+ type Value = DecodeResult<Datum<'static>>;
+
+ fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str(complex_expected(self.data_type))
+ }
+
+ fn visit_unit<E>(self) -> Result<Self::Value, E> {
+ Ok(self.null())
+ }
+
+ fn visit_bool<E>(self, _value: bool) -> Result<Self::Value, E> {
+ Ok(self.type_error("a boolean"))
+ }
+
+ fn visit_i64<E>(self, _value: i64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_u64<E>(self, _value: u64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_f64<E>(self, _value: f64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_str<E>(self, _value: &str) -> Result<Self::Value, E> {
+ Ok(self.type_error("a string"))
+ }
+
+ fn visit_seq<A>(self, sequence: A) -> Result<Self::Value, A::Error>
+ where
+ A: SeqAccess<'de>,
+ {
+ match self.data_type {
+ DataType::Array(data_type) => {
+ decode_array_sequence(sequence, &self.path, data_type.get_element_type())
+ }
+ DataType::Map(data_type) => decode_map_sequence(
+ sequence,
+ &self.path,
+ data_type.key_type(),
+ data_type.value_type(),
+ ),
+ DataType::Row(_) => {
+ let mut sequence = sequence;
+ drain_sequence(&mut sequence)?;
+ Ok(self.type_error("an array"))
+ }
+ _ => unreachable!("complex visitor receives only ARRAY, MAP, or ROW"),
+ }
+ }
+
+ fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>
+ where
+ A: MapAccess<'de>,
+ {
+ match self.data_type {
+ DataType::Row(row_type) => {
+ let row = decode_row_map(map, row_type, RowContext::Nested { path: self.path })?;
+ Ok(row.map(|row| Datum::Row(Box::new(row))))
+ }
+ DataType::Array(_) | DataType::Map(_) => {
+ let mut map = map;
+ drain_map(&mut map)?;
+ Ok(self.type_error("an object"))
+ }
+ _ => unreachable!("complex visitor receives only ARRAY, MAP, or ROW"),
+ }
+ }
+}
+
+fn complex_expected(data_type: &DataType) -> &'static str {
+ match data_type {
+ DataType::Array(_) => "ARRAY (a JSON array)",
+ DataType::Map(_) => "MAP (an array of {key, value} objects)",
+ DataType::Row(_) => "ROW (a JSON object of its declared fields)",
+ _ => unreachable!("only complex data types have a container expectation"),
+ }
+}
+
+fn drain_sequence<'de, A>(sequence: &mut A) -> Result<(), A::Error>
+where
+ A: SeqAccess<'de>,
+{
+ while sequence.next_element::<IgnoredAny>()?.is_some() {}
+ Ok(())
+}
+
+fn drain_map<'de, A>(map: &mut A) -> Result<(), A::Error>
+where
+ A: MapAccess<'de>,
+{
+ while map.next_entry::<IgnoredAny, IgnoredAny>()?.is_some() {}
+ Ok(())
+}
+
+fn decode_scalar(
+ path: &ValuePath<'_>,
+ data_type: &DataType,
+ value: &ScalarValue<'_>,
+) -> DecodeResult<Datum<'static>> {
+ if matches!(value, ScalarValue::Null) {
+ return if data_type.is_nullable() {
+ Ok(Datum::Null)
+ } else {
+ Err(path.invalid("must not be null"))
+ };
+ }
+ match data_type {
+ DataType::Boolean(_) => match value {
+ ScalarValue::Boolean(parsed) => Ok(Datum::Bool(*parsed)),
+ _ => Err(path.type_error("BOOLEAN (a JSON boolean)", value.kind())),
+ },
+ DataType::TinyInt(_) => {
+ decode_integer(path, value, "TINYINT", i8::MIN as i64, i8::MAX as i64)
+ .map(|parsed| Datum::Int8(parsed as i8))
+ }
+ DataType::SmallInt(_) => {
+ decode_integer(path, value, "SMALLINT", i16::MIN as i64, i16::MAX as i64)
+ .map(|parsed| Datum::Int16(parsed as i16))
+ }
+ DataType::Int(_) => decode_integer(path, value, "INT", i32::MIN as i64, i32::MAX as i64)
+ .map(|parsed| Datum::Int32(parsed as i32)),
+ DataType::BigInt(_) => decode_bigint(path, value).map(Datum::Int64),
+ DataType::Float(_) => decode_float32(path, value),
+ DataType::Double(_) => decode_float(path, value, "DOUBLE").map(Datum::from),
+ DataType::Char(_) | DataType::String(_) => match value {
+ ScalarValue::String(text) => Ok(Datum::String(Cow::Owned(text.clone()))),
+ _ => Err(path.type_error("STRING (a JSON string)", value.kind())),
+ },
+ DataType::Decimal(data_type) => {
+ decode_decimal(path, value, data_type.precision(), data_type.scale())
+ }
+ DataType::Bytes(_) => decode_binary(path, value, None),
+ DataType::Binary(data_type) => decode_binary(path, value, Some(data_type.length())),
+ DataType::Date(_) => decode_date(path, value),
+ DataType::Time(data_type) => decode_time(path, value, data_type.precision()),
+ DataType::Timestamp(data_type) => {
+ decode_timestamp(path, value, data_type.precision(), false)
+ }
+ DataType::TimestampLTz(data_type) => {
+ decode_timestamp(path, value, data_type.precision(), true)
+ }
+ DataType::Array(_) | DataType::Map(_) | DataType::Row(_) => {
+ unreachable!("container values use the complex visitor")
+ }
+ }
+}
+fn decode_integer(
+ path: &ValuePath<'_>,
+ value: &ScalarValue<'_>,
+ type_name: &str,
+ min: i64,
+ max: i64,
+) -> Result<i64, RowDecodeError> {
+ let expected = format!("{type_name} (an integer in [{min}, {max}])");
Review Comment:
You're right. We were aligned with ProtoJSON's canonical output forms, but not with its parser acceptance rules.
Integer types now accept quoted and unquoted values, including exponent notation and zero fractional parts when the exact result is integral. FLOAT and DOUBLE also accept finite numeric strings. Integer decoding remains exact and range-checked without going through f64.
##########
fluss-gateway/src/protocol/rest/codec/decoder.rs:
##########
@@ -0,0 +1,1927 @@
+// 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.
+
+//! Schema-aware streaming conversion from one raw FIP-49 JSON row to [`GenericRow`].
+//!
+//! Serde owns JSON syntax and container traversal. Schema-matching ARRAY, MAP, and ROW values
+//! recurse only along the validated [`RowType`], bounded by [`MAX_TYPE_NESTING`]. Scalar and
+//! ignored subtrees use serde_json's iterative [`RawValue`] and [`IgnoredAny`] scan path, so JSON
+//! depth does not grow this decoder's call stack. No container subtree is reparsed, keeping work
+//! linear in row bytes apart from local scalar decoding; the HTTP layer separately bounds body
+//! size. Recheck the iterative-scan assumption when upgrading serde_json.
+
+use crate::error::GatewayError;
+use crate::protocol::rest::codec::temporal::{parse_date, parse_time, parse_timestamp};
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64;
+use fluss::metadata::{DataType, RowType};
+use fluss::row::{
+ Date, Datum, Decimal, FlussArrayWriter, FlussMapWriter, GenericRow, Time, TimestampLtz,
+ TimestampNtz,
+};
+use serde::Deserialize;
+use serde::de::{self, DeserializeSeed, IgnoredAny, MapAccess, SeqAccess, Visitor};
+use serde_json::value::RawValue;
+use std::borrow::Cow;
+use std::collections::{HashMap, HashSet};
+use std::fmt;
+use std::sync::Arc;
+
+/// Bounds schema-recursive validation and decoding; it is not a JSON body-depth limit.
+const MAX_TYPE_NESTING: usize = 64;
+const NANOS_PER_MILLI: i64 = 1_000_000;
+const MILLIS_PER_SECOND: i64 = 1_000;
+const MILLIS_PER_DAY: i64 = 86_400_000;
+
+/// Whether a row supplies a complete mutation or only selected fields.
+#[derive(Debug, Clone, Copy)]
+#[allow(dead_code)]
+pub(crate) enum RowShape<'a> {
+ /// Every non-nullable column must be present. Missing nullable columns become null.
+ Complete,
+ /// Only the named columns must be present and non-null. Other missing columns become null.
+ Sparse(&'a [String]),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum RowDecodeErrorKind {
+ SchemaMismatch,
+ InvalidValue,
+}
+
+/// One row decode failure plus whether refreshed metadata could plausibly resolve it.
+#[derive(Debug)]
+pub(crate) struct RowDecodeError {
+ kind: RowDecodeErrorKind,
+ error: GatewayError,
+}
+
+#[allow(dead_code)]
+impl RowDecodeError {
+ fn schema_mismatch(error: GatewayError) -> Self {
+ Self {
+ kind: RowDecodeErrorKind::SchemaMismatch,
+ error,
+ }
+ }
+
+ fn invalid(error: GatewayError) -> Self {
+ Self {
+ kind: RowDecodeErrorKind::InvalidValue,
+ error,
+ }
+ }
+
+ /// Returns true only for failures that a newer table schema may resolve.
+ pub(crate) fn is_schema_mismatch(&self) -> bool {
+ self.kind == RowDecodeErrorKind::SchemaMismatch
+ }
+
+ /// Returns the client-safe error message.
+ pub(crate) fn message(&self) -> &str {
+ self.error.message()
+ }
+
+ /// Converts this failure into the standard Gateway error envelope input.
+ pub(crate) fn into_gateway_error(self) -> GatewayError {
+ self.error
+ }
+}
+
+impl From<GatewayError> for RowDecodeError {
+ fn from(error: GatewayError) -> Self {
+ Self::invalid(error)
+ }
+}
+
+struct SchemaDecoderInner {
+ row_type: RowType,
+ column_indexes: HashMap<String, usize>,
+}
+
+/// Immutable reusable decoder compiled from one Fluss row type.
+#[derive(Clone)]
+#[allow(dead_code)]
+pub(crate) struct SchemaDecoder {
+ inner: Arc<SchemaDecoderInner>,
+}
+
+impl fmt::Debug for SchemaDecoder {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter
+ .debug_struct("SchemaDecoder")
+ .field("row_type", &self.inner.row_type)
+ .finish()
+ }
+}
+
+#[allow(dead_code)]
+impl SchemaDecoder {
+ /// Builds a decoder for a complete table row type.
+ pub(crate) fn new(row_type: RowType) -> Result<Self, GatewayError> {
+ validate_row_type(&row_type, 0)?;
+ if row_type.fields().is_empty() {
+ return Err(GatewayError::internal(
+ "cannot build a row decoder for an empty table schema",
+ ));
+ }
+ let column_indexes = row_type
+ .fields()
+ .iter()
+ .enumerate()
+ .map(|(index, field)| (field.name.clone(), index))
+ .collect();
+ Ok(Self {
+ inner: Arc::new(SchemaDecoderInner {
+ row_type,
+ column_indexes,
+ }),
+ })
+ }
+
+ /// Returns the exact row type this decoder was built from.
+ pub(crate) fn row_type(&self) -> &RowType {
+ &self.inner.row_type
+ }
+
+ /// Decodes one raw JSON object into complete schema order.
+ pub(crate) fn decode_row(
+ &self,
+ label: &str,
+ row_json: &[u8],
+ shape: RowShape<'_>,
+ ) -> Result<GenericRow<'static>, RowDecodeError> {
+ let required = self.required_columns(label, shape)?;
+ let seed = RootRowSeed {
+ label,
+ row_type: &self.inner.row_type,
+ column_indexes: &self.inner.column_indexes,
+ required: &required,
+ };
+ let mut deserializer = serde_json::Deserializer::from_slice(row_json);
+ let decoded = seed
+ .deserialize(&mut deserializer)
+ .map_err(|error| invalid_json_error(label, error))?;
+ deserializer
+ .end()
+ .map_err(|error| invalid_json_error(label, error))?;
+ decoded
+ }
+
+ fn required_columns(
+ &self,
+ label: &str,
+ shape: RowShape<'_>,
+ ) -> Result<Vec<bool>, RowDecodeError> {
+ match shape {
+ RowShape::Complete => Ok(self
+ .inner
+ .row_type
+ .fields()
+ .iter()
+ .map(|field| !field.data_type.is_nullable())
+ .collect()),
+ RowShape::Sparse(required_columns) => {
+ let mut required = vec![false; self.inner.row_type.fields().len()];
+ for column in required_columns {
+ let Some(index) = self.inner.column_indexes.get(column) else {
+ return Err(RowDecodeError::schema_mismatch(
+ GatewayError::invalid_argument(format!(
+ "{label}: required column `{column}` is not part of the table schema"
+ )),
+ ));
+ };
+ required[*index] = true;
+ }
+ Ok(required)
+ }
+ }
+ }
+}
+
+fn invalid_json_error(label: &str, error: serde_json::Error) -> RowDecodeError {
+ RowDecodeError::invalid(GatewayError::invalid_argument(format!(
+ "{label}: invalid JSON row: {error}"
+ )))
+}
+
+fn validate_row_type(row_type: &RowType, depth: usize) -> Result<(), GatewayError> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::internal(format!(
+ "row type nesting exceeds {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ let mut names = HashSet::with_capacity(row_type.fields().len());
+ for field in row_type.fields() {
+ if field.name.is_empty() {
+ return Err(GatewayError::internal(
+ "row type contains an empty field name",
+ ));
+ }
+ if !names.insert(field.name.as_str()) {
+ return Err(GatewayError::internal(format!(
+ "row type contains duplicate field `{}`",
+ field.name
+ )));
+ }
+ validate_data_type(&field.data_type, depth + 1)?;
+ }
+ Ok(())
+}
+
+fn validate_data_type(data_type: &DataType, depth: usize) -> Result<(), GatewayError> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::internal(format!(
+ "row type nesting exceeds {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ match data_type {
+ DataType::Char(data_type) if data_type.length() == 0 => {
+ Err(GatewayError::internal("CHAR length must be at least one"))
+ }
+ DataType::Binary(data_type) if data_type.length() == 0 => {
+ Err(GatewayError::internal("BINARY length must be at least one"))
+ }
+ DataType::Array(data_type) => validate_data_type(data_type.get_element_type(), depth + 1),
+ DataType::Map(data_type) => {
+ if data_type.key_type().is_nullable() {
+ return Err(GatewayError::internal("MAP key type must not be nullable"));
+ }
+ if is_complex(data_type.key_type()) {
+ return Err(GatewayError::invalid_argument(
+ "FIP-49 JSON does not support ARRAY, MAP, or ROW values as MAP keys",
Review Comment:
Good point. The original behavior followed the current FIP-49 specification, whic requires 400 for complex MAP keys. However, I agree that this is a schema capability limitation rather than an invalid row body, so 501 is more appropriate.
I’ve updated the decoder to return `Unsupported` and removed FIP-49 references from client-facing error messages.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FIP-49][gateway] Add schema-aware JSON row decoder [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni merged PR #4104:
URL: https://github.com/apache/fluss/pull/4104
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FIP-49][gateway] Add schema-aware JSON and Arrow row codecs for FIP-49 data types [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #4104:
URL: https://github.com/apache/fluss/pull/4104#discussion_r3863797968
##########
fluss-gateway/src/protocol/rest/codec/decoder.rs:
##########
@@ -0,0 +1,1927 @@
+// 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.
+
+//! Schema-aware streaming conversion from one raw FIP-49 JSON row to [`GenericRow`].
+//!
+//! Serde owns JSON syntax and container traversal. Schema-matching ARRAY, MAP, and ROW values
+//! recurse only along the validated [`RowType`], bounded by [`MAX_TYPE_NESTING`]. Scalar and
+//! ignored subtrees use serde_json's iterative [`RawValue`] and [`IgnoredAny`] scan path, so JSON
+//! depth does not grow this decoder's call stack. No container subtree is reparsed, keeping work
+//! linear in row bytes apart from local scalar decoding; the HTTP layer separately bounds body
+//! size. Recheck the iterative-scan assumption when upgrading serde_json.
+
+use crate::error::GatewayError;
+use crate::protocol::rest::codec::temporal::{parse_date, parse_time, parse_timestamp};
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64;
+use fluss::metadata::{DataType, RowType};
+use fluss::row::{
+ Date, Datum, Decimal, FlussArrayWriter, FlussMapWriter, GenericRow, Time, TimestampLtz,
+ TimestampNtz,
+};
+use serde::Deserialize;
+use serde::de::{self, DeserializeSeed, IgnoredAny, MapAccess, SeqAccess, Visitor};
+use serde_json::value::RawValue;
+use std::borrow::Cow;
+use std::collections::{HashMap, HashSet};
+use std::fmt;
+use std::sync::Arc;
+
+/// Bounds schema-recursive validation and decoding; it is not a JSON body-depth limit.
+const MAX_TYPE_NESTING: usize = 64;
+const NANOS_PER_MILLI: i64 = 1_000_000;
+const MILLIS_PER_SECOND: i64 = 1_000;
+const MILLIS_PER_DAY: i64 = 86_400_000;
+
+/// Whether a row supplies a complete mutation or only selected fields.
+#[derive(Debug, Clone, Copy)]
+#[allow(dead_code)]
+pub(crate) enum RowShape<'a> {
+ /// Every non-nullable column must be present. Missing nullable columns become null.
+ Complete,
+ /// Only the named columns must be present and non-null. Other missing columns become null.
+ Sparse(&'a [String]),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum RowDecodeErrorKind {
+ SchemaMismatch,
+ InvalidValue,
+}
+
+/// One row decode failure plus whether refreshed metadata could plausibly resolve it.
+#[derive(Debug)]
+pub(crate) struct RowDecodeError {
+ kind: RowDecodeErrorKind,
+ error: GatewayError,
+}
+
+#[allow(dead_code)]
+impl RowDecodeError {
+ fn schema_mismatch(error: GatewayError) -> Self {
+ Self {
+ kind: RowDecodeErrorKind::SchemaMismatch,
+ error,
+ }
+ }
+
+ fn invalid(error: GatewayError) -> Self {
+ Self {
+ kind: RowDecodeErrorKind::InvalidValue,
+ error,
+ }
+ }
+
+ /// Returns true only for failures that a newer table schema may resolve.
+ pub(crate) fn is_schema_mismatch(&self) -> bool {
+ self.kind == RowDecodeErrorKind::SchemaMismatch
+ }
+
+ /// Returns the client-safe error message.
+ pub(crate) fn message(&self) -> &str {
+ self.error.message()
+ }
+
+ /// Converts this failure into the standard Gateway error envelope input.
+ pub(crate) fn into_gateway_error(self) -> GatewayError {
+ self.error
+ }
+}
+
+impl From<GatewayError> for RowDecodeError {
+ fn from(error: GatewayError) -> Self {
+ Self::invalid(error)
+ }
+}
+
+struct SchemaDecoderInner {
+ row_type: RowType,
+ column_indexes: HashMap<String, usize>,
+}
+
+/// Immutable reusable decoder compiled from one Fluss row type.
+#[derive(Clone)]
+#[allow(dead_code)]
+pub(crate) struct SchemaDecoder {
+ inner: Arc<SchemaDecoderInner>,
+}
+
+impl fmt::Debug for SchemaDecoder {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter
+ .debug_struct("SchemaDecoder")
+ .field("row_type", &self.inner.row_type)
+ .finish()
+ }
+}
+
+#[allow(dead_code)]
+impl SchemaDecoder {
+ /// Builds a decoder for a complete table row type.
+ pub(crate) fn new(row_type: RowType) -> Result<Self, GatewayError> {
+ validate_row_type(&row_type, 0)?;
+ if row_type.fields().is_empty() {
+ return Err(GatewayError::internal(
+ "cannot build a row decoder for an empty table schema",
+ ));
+ }
+ let column_indexes = row_type
+ .fields()
+ .iter()
+ .enumerate()
+ .map(|(index, field)| (field.name.clone(), index))
+ .collect();
+ Ok(Self {
+ inner: Arc::new(SchemaDecoderInner {
+ row_type,
+ column_indexes,
+ }),
+ })
+ }
+
+ /// Returns the exact row type this decoder was built from.
+ pub(crate) fn row_type(&self) -> &RowType {
+ &self.inner.row_type
+ }
+
+ /// Decodes one raw JSON object into complete schema order.
+ pub(crate) fn decode_row(
+ &self,
+ label: &str,
+ row_json: &[u8],
+ shape: RowShape<'_>,
+ ) -> Result<GenericRow<'static>, RowDecodeError> {
+ let required = self.required_columns(label, shape)?;
+ let seed = RootRowSeed {
+ label,
+ row_type: &self.inner.row_type,
+ column_indexes: &self.inner.column_indexes,
+ required: &required,
+ };
+ let mut deserializer = serde_json::Deserializer::from_slice(row_json);
+ let decoded = seed
+ .deserialize(&mut deserializer)
+ .map_err(|error| invalid_json_error(label, error))?;
+ deserializer
+ .end()
+ .map_err(|error| invalid_json_error(label, error))?;
+ decoded
+ }
+
+ fn required_columns(
+ &self,
+ label: &str,
+ shape: RowShape<'_>,
+ ) -> Result<Vec<bool>, RowDecodeError> {
+ match shape {
+ RowShape::Complete => Ok(self
+ .inner
+ .row_type
+ .fields()
+ .iter()
+ .map(|field| !field.data_type.is_nullable())
+ .collect()),
+ RowShape::Sparse(required_columns) => {
+ let mut required = vec![false; self.inner.row_type.fields().len()];
+ for column in required_columns {
+ let Some(index) = self.inner.column_indexes.get(column) else {
+ return Err(RowDecodeError::schema_mismatch(
+ GatewayError::invalid_argument(format!(
+ "{label}: required column `{column}` is not part of the table schema"
+ )),
+ ));
+ };
+ required[*index] = true;
+ }
+ Ok(required)
+ }
+ }
+ }
+}
+
+fn invalid_json_error(label: &str, error: serde_json::Error) -> RowDecodeError {
+ RowDecodeError::invalid(GatewayError::invalid_argument(format!(
+ "{label}: invalid JSON row: {error}"
+ )))
+}
+
+fn validate_row_type(row_type: &RowType, depth: usize) -> Result<(), GatewayError> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::internal(format!(
+ "row type nesting exceeds {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ let mut names = HashSet::with_capacity(row_type.fields().len());
+ for field in row_type.fields() {
+ if field.name.is_empty() {
+ return Err(GatewayError::internal(
+ "row type contains an empty field name",
+ ));
+ }
+ if !names.insert(field.name.as_str()) {
+ return Err(GatewayError::internal(format!(
+ "row type contains duplicate field `{}`",
+ field.name
+ )));
+ }
+ validate_data_type(&field.data_type, depth + 1)?;
+ }
+ Ok(())
+}
+
+fn validate_data_type(data_type: &DataType, depth: usize) -> Result<(), GatewayError> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::internal(format!(
+ "row type nesting exceeds {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ match data_type {
+ DataType::Char(data_type) if data_type.length() == 0 => {
+ Err(GatewayError::internal("CHAR length must be at least one"))
+ }
+ DataType::Binary(data_type) if data_type.length() == 0 => {
+ Err(GatewayError::internal("BINARY length must be at least one"))
+ }
+ DataType::Array(data_type) => validate_data_type(data_type.get_element_type(), depth + 1),
+ DataType::Map(data_type) => {
+ if data_type.key_type().is_nullable() {
+ return Err(GatewayError::internal("MAP key type must not be nullable"));
+ }
+ if is_complex(data_type.key_type()) {
+ return Err(GatewayError::invalid_argument(
+ "FIP-49 JSON does not support ARRAY, MAP, or ROW values as MAP keys",
Review Comment:
A MAP with a complex key fails when the decoder is built, so every write to such a table gets 400 for a body that was fine. Is invalid_argument right when the caller can't change the schema?
And we better remove FIP-49 references - we shoudn't expect users to go there.
##########
fluss-gateway/src/protocol/rest/codec/temporal.rs:
##########
@@ -0,0 +1,252 @@
+// 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.
+
+//! Shared temporal parsing and formatting for both codec directions.
+
+const DAYS_PER_GREGORIAN_CYCLE: i64 = 146_097;
+const DAYS_FROM_CIVIL_EPOCH_TO_UNIX_EPOCH: i64 = 719_468;
+
+pub(super) struct ParsedTimestamp {
+ pub(super) days: i64,
+ pub(super) seconds_of_day: i64,
+ pub(super) frac_nanos: i64,
+ pub(super) offset_seconds: Option<i32>,
+}
+
+pub(super) fn parse_date(text: &str) -> Option<i64> {
+ let (negative, unsigned) = match text.strip_prefix('-') {
+ Some(rest) => (true, rest),
+ None => (false, text.strip_prefix('+').unwrap_or(text)),
+ };
+ let (year, month_and_day) = unsigned.split_once('-')?;
+ let (month, day) = month_and_day.split_once('-')?;
+ if year.len() < 4
+ || month.len() != 2
+ || day.len() != 2
+ || month_and_day.matches('-').count() != 1
+ {
+ return None;
+ }
+ let unsigned_year: i64 = digits_only(year)?.parse().ok()?;
+ let year = if negative {
+ unsigned_year.checked_neg()?
+ } else {
+ unsigned_year
+ };
+ let month: u32 = digits_only(month)?.parse().ok()?;
+ let day: u32 = digits_only(day)?.parse().ok()?;
+ if !(1..=12).contains(&month) || !(1..=days_in_month(year, month)).contains(&day) {
+ return None;
+ }
+ days_from_civil(year, month, day)
+}
+
+pub(super) fn parse_time(text: &str) -> Option<(i64, i64)> {
+ let (clock, fraction) = match text.split_once('.') {
+ Some((clock, fraction)) => (clock, Some(fraction)),
+ None => (text, None),
+ };
+ let bytes = clock.as_bytes();
+ let (hour, minute, second) = match bytes.len() {
+ 5 if bytes[2] == b':' => (clock.get(0..2)?, clock.get(3..5)?, "0"),
+ 8 if bytes[2] == b':' && bytes[5] == b':' => {
+ (clock.get(0..2)?, clock.get(3..5)?, clock.get(6..8)?)
+ }
+ _ => return None,
+ };
+ let hour: i64 = digits_only(hour)?.parse().ok()?;
+ let minute: i64 = digits_only(minute)?.parse().ok()?;
+ let second: i64 = digits_only(second)?.parse().ok()?;
+ if hour > 23 || minute > 59 || second > 59 {
+ return None;
+ }
+ let frac_nanos = match fraction {
+ None => 0,
+ Some(fraction) => {
+ if fraction.is_empty() || fraction.len() > 9 {
+ return None;
+ }
+ let digits: i64 = digits_only(fraction)?.parse().ok()?;
+ digits * 10_i64.pow(9 - fraction.len() as u32)
+ }
+ };
+ Some((hour * 3_600 + minute * 60 + second, frac_nanos))
+}
+
+pub(super) fn parse_timestamp(text: &str) -> Option<ParsedTimestamp> {
+ let separator = text.find(['T', ' '])?;
Review Comment:
A space works in place of T, so "2026-01-01 00:00:00" parses. Intended? Shall we specify this in some docs?
##########
fluss-gateway/src/protocol/rest/codec/decoder.rs:
##########
@@ -0,0 +1,1927 @@
+// 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.
+
+//! Schema-aware streaming conversion from one raw FIP-49 JSON row to [`GenericRow`].
+//!
+//! Serde owns JSON syntax and container traversal. Schema-matching ARRAY, MAP, and ROW values
+//! recurse only along the validated [`RowType`], bounded by [`MAX_TYPE_NESTING`]. Scalar and
+//! ignored subtrees use serde_json's iterative [`RawValue`] and [`IgnoredAny`] scan path, so JSON
+//! depth does not grow this decoder's call stack. No container subtree is reparsed, keeping work
+//! linear in row bytes apart from local scalar decoding; the HTTP layer separately bounds body
+//! size. Recheck the iterative-scan assumption when upgrading serde_json.
+
+use crate::error::GatewayError;
+use crate::protocol::rest::codec::temporal::{parse_date, parse_time, parse_timestamp};
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64;
+use fluss::metadata::{DataType, RowType};
+use fluss::row::{
+ Date, Datum, Decimal, FlussArrayWriter, FlussMapWriter, GenericRow, Time, TimestampLtz,
+ TimestampNtz,
+};
+use serde::Deserialize;
+use serde::de::{self, DeserializeSeed, IgnoredAny, MapAccess, SeqAccess, Visitor};
+use serde_json::value::RawValue;
+use std::borrow::Cow;
+use std::collections::{HashMap, HashSet};
+use std::fmt;
+use std::sync::Arc;
+
+/// Bounds schema-recursive validation and decoding; it is not a JSON body-depth limit.
+const MAX_TYPE_NESTING: usize = 64;
+const NANOS_PER_MILLI: i64 = 1_000_000;
+const MILLIS_PER_SECOND: i64 = 1_000;
+const MILLIS_PER_DAY: i64 = 86_400_000;
+
+/// Whether a row supplies a complete mutation or only selected fields.
+#[derive(Debug, Clone, Copy)]
+#[allow(dead_code)]
+pub(crate) enum RowShape<'a> {
+ /// Every non-nullable column must be present. Missing nullable columns become null.
+ Complete,
+ /// Only the named columns must be present and non-null. Other missing columns become null.
+ Sparse(&'a [String]),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum RowDecodeErrorKind {
+ SchemaMismatch,
+ InvalidValue,
+}
+
+/// One row decode failure plus whether refreshed metadata could plausibly resolve it.
+#[derive(Debug)]
+pub(crate) struct RowDecodeError {
+ kind: RowDecodeErrorKind,
+ error: GatewayError,
+}
+
+#[allow(dead_code)]
+impl RowDecodeError {
+ fn schema_mismatch(error: GatewayError) -> Self {
+ Self {
+ kind: RowDecodeErrorKind::SchemaMismatch,
+ error,
+ }
+ }
+
+ fn invalid(error: GatewayError) -> Self {
+ Self {
+ kind: RowDecodeErrorKind::InvalidValue,
+ error,
+ }
+ }
+
+ /// Returns true only for failures that a newer table schema may resolve.
+ pub(crate) fn is_schema_mismatch(&self) -> bool {
+ self.kind == RowDecodeErrorKind::SchemaMismatch
+ }
+
+ /// Returns the client-safe error message.
+ pub(crate) fn message(&self) -> &str {
+ self.error.message()
+ }
+
+ /// Converts this failure into the standard Gateway error envelope input.
+ pub(crate) fn into_gateway_error(self) -> GatewayError {
+ self.error
+ }
+}
+
+impl From<GatewayError> for RowDecodeError {
+ fn from(error: GatewayError) -> Self {
+ Self::invalid(error)
+ }
+}
+
+struct SchemaDecoderInner {
+ row_type: RowType,
+ column_indexes: HashMap<String, usize>,
+}
+
+/// Immutable reusable decoder compiled from one Fluss row type.
+#[derive(Clone)]
+#[allow(dead_code)]
+pub(crate) struct SchemaDecoder {
+ inner: Arc<SchemaDecoderInner>,
+}
+
+impl fmt::Debug for SchemaDecoder {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter
+ .debug_struct("SchemaDecoder")
+ .field("row_type", &self.inner.row_type)
+ .finish()
+ }
+}
+
+#[allow(dead_code)]
+impl SchemaDecoder {
+ /// Builds a decoder for a complete table row type.
+ pub(crate) fn new(row_type: RowType) -> Result<Self, GatewayError> {
+ validate_row_type(&row_type, 0)?;
+ if row_type.fields().is_empty() {
+ return Err(GatewayError::internal(
+ "cannot build a row decoder for an empty table schema",
+ ));
+ }
+ let column_indexes = row_type
+ .fields()
+ .iter()
+ .enumerate()
+ .map(|(index, field)| (field.name.clone(), index))
+ .collect();
+ Ok(Self {
+ inner: Arc::new(SchemaDecoderInner {
+ row_type,
+ column_indexes,
+ }),
+ })
+ }
+
+ /// Returns the exact row type this decoder was built from.
+ pub(crate) fn row_type(&self) -> &RowType {
+ &self.inner.row_type
+ }
+
+ /// Decodes one raw JSON object into complete schema order.
+ pub(crate) fn decode_row(
+ &self,
+ label: &str,
+ row_json: &[u8],
+ shape: RowShape<'_>,
+ ) -> Result<GenericRow<'static>, RowDecodeError> {
+ let required = self.required_columns(label, shape)?;
+ let seed = RootRowSeed {
+ label,
+ row_type: &self.inner.row_type,
+ column_indexes: &self.inner.column_indexes,
+ required: &required,
+ };
+ let mut deserializer = serde_json::Deserializer::from_slice(row_json);
+ let decoded = seed
+ .deserialize(&mut deserializer)
+ .map_err(|error| invalid_json_error(label, error))?;
+ deserializer
+ .end()
+ .map_err(|error| invalid_json_error(label, error))?;
+ decoded
+ }
+
+ fn required_columns(
+ &self,
+ label: &str,
+ shape: RowShape<'_>,
+ ) -> Result<Vec<bool>, RowDecodeError> {
+ match shape {
+ RowShape::Complete => Ok(self
+ .inner
+ .row_type
+ .fields()
+ .iter()
+ .map(|field| !field.data_type.is_nullable())
+ .collect()),
+ RowShape::Sparse(required_columns) => {
+ let mut required = vec![false; self.inner.row_type.fields().len()];
+ for column in required_columns {
+ let Some(index) = self.inner.column_indexes.get(column) else {
+ return Err(RowDecodeError::schema_mismatch(
+ GatewayError::invalid_argument(format!(
+ "{label}: required column `{column}` is not part of the table schema"
+ )),
+ ));
+ };
+ required[*index] = true;
+ }
+ Ok(required)
+ }
+ }
+ }
+}
+
+fn invalid_json_error(label: &str, error: serde_json::Error) -> RowDecodeError {
+ RowDecodeError::invalid(GatewayError::invalid_argument(format!(
+ "{label}: invalid JSON row: {error}"
+ )))
+}
+
+fn validate_row_type(row_type: &RowType, depth: usize) -> Result<(), GatewayError> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::internal(format!(
+ "row type nesting exceeds {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ let mut names = HashSet::with_capacity(row_type.fields().len());
+ for field in row_type.fields() {
+ if field.name.is_empty() {
+ return Err(GatewayError::internal(
+ "row type contains an empty field name",
+ ));
+ }
+ if !names.insert(field.name.as_str()) {
+ return Err(GatewayError::internal(format!(
+ "row type contains duplicate field `{}`",
+ field.name
+ )));
+ }
+ validate_data_type(&field.data_type, depth + 1)?;
+ }
+ Ok(())
+}
+
+fn validate_data_type(data_type: &DataType, depth: usize) -> Result<(), GatewayError> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::internal(format!(
+ "row type nesting exceeds {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ match data_type {
+ DataType::Char(data_type) if data_type.length() == 0 => {
+ Err(GatewayError::internal("CHAR length must be at least one"))
+ }
+ DataType::Binary(data_type) if data_type.length() == 0 => {
+ Err(GatewayError::internal("BINARY length must be at least one"))
+ }
+ DataType::Array(data_type) => validate_data_type(data_type.get_element_type(), depth + 1),
+ DataType::Map(data_type) => {
+ if data_type.key_type().is_nullable() {
+ return Err(GatewayError::internal("MAP key type must not be nullable"));
+ }
+ if is_complex(data_type.key_type()) {
+ return Err(GatewayError::invalid_argument(
+ "FIP-49 JSON does not support ARRAY, MAP, or ROW values as MAP keys",
+ ));
+ }
+ validate_data_type(data_type.key_type(), depth + 1)?;
+ validate_data_type(data_type.value_type(), depth + 1)
+ }
+ DataType::Row(row_type) => validate_row_type(row_type, depth + 1),
+ _ => Ok(()),
+ }
+}
+
+fn is_complex(data_type: &DataType) -> bool {
+ matches!(
+ data_type,
+ DataType::Array(_) | DataType::Map(_) | DataType::Row(_)
+ )
+}
+
+#[derive(Clone)]
+struct ValuePath<'a> {
+ label: &'a str,
+ path: String,
+}
+
+impl<'a> ValuePath<'a> {
+ fn column(label: &'a str, column: &str) -> Self {
+ Self {
+ label,
+ path: column.to_string(),
+ }
+ }
+
+ fn element(&self, index: usize) -> Self {
+ self.nested(format!("{}[{index}]", self.path))
+ }
+
+ fn nested_field(&self, name: &str) -> Self {
+ self.nested(format!("{}.{name}", self.path))
+ }
+
+ fn map_part(&self, index: usize, part: &str) -> Self {
+ self.nested(format!("{}[{index}].{part}", self.path))
+ }
+
+ fn nested(&self, path: String) -> Self {
+ Self {
+ label: self.label,
+ path,
+ }
+ }
+
+ fn invalid(&self, reason: impl fmt::Display) -> RowDecodeError {
+ RowDecodeError::invalid(GatewayError::invalid_argument(format!(
+ "{}: column `{}` {reason}",
+ self.label, self.path
+ )))
+ }
+
+ fn schema_mismatch(&self, reason: impl fmt::Display) -> RowDecodeError {
+ RowDecodeError::schema_mismatch(GatewayError::invalid_argument(format!(
+ "{}: column `{}` {reason}",
+ self.label, self.path
+ )))
+ }
+
+ fn type_error(&self, expected: &str, actual: &str) -> RowDecodeError {
+ self.invalid(format!("expects {expected}, got {actual}"))
+ }
+}
+
+type DecodeResult<T> = Result<T, RowDecodeError>;
+
+enum ScalarValue<'a> {
+ Null,
+ Boolean(bool),
+ ExactNumber(&'a str),
+ String(String),
+ Array,
+ Object,
+}
+
+impl ScalarValue<'_> {
+ fn kind(&self) -> &'static str {
+ match self {
+ Self::Null => "null",
+ Self::Boolean(_) => "a boolean",
+ Self::ExactNumber(_) => "a number",
+ Self::String(_) => "a string",
+ Self::Array => "an array",
+ Self::Object => "an object",
+ }
+ }
+}
+
+fn parse_scalar_value(raw: &RawValue) -> Result<ScalarValue<'_>, serde_json::Error> {
+ let text = raw.get().trim();
+ match text.as_bytes().first().copied() {
+ Some(b'n') => Ok(ScalarValue::Null),
+ Some(b't') => Ok(ScalarValue::Boolean(true)),
+ Some(b'f') => Ok(ScalarValue::Boolean(false)),
+ Some(b'"') => serde_json::from_str(text).map(ScalarValue::String),
+ Some(b'-' | b'0'..=b'9') => Ok(ScalarValue::ExactNumber(text)),
+ Some(b'[') => Ok(ScalarValue::Array),
+ Some(b'{') => Ok(ScalarValue::Object),
+ _ => unreachable!("RawValue always contains one complete JSON value"),
+ }
+}
+
+struct RootRowSeed<'a> {
+ label: &'a str,
+ row_type: &'a RowType,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+}
+
+impl<'de> DeserializeSeed<'de> for RootRowSeed<'_> {
+ type Value = DecodeResult<GenericRow<'static>>;
+
+ fn deserialize<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
+ where
+ D: de::Deserializer<'de>,
+ {
+ deserializer.deserialize_any(RootRowVisitor {
+ label: self.label,
+ row_type: self.row_type,
+ column_indexes: self.column_indexes,
+ required: self.required,
+ })
+ }
+}
+
+struct RootRowVisitor<'a> {
+ label: &'a str,
+ row_type: &'a RowType,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+}
+
+impl RootRowVisitor<'_> {
+ fn type_error(&self, actual: &str) -> DecodeResult<GenericRow<'static>> {
+ Err(RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{}: row must be a JSON object, got {actual}", self.label),
+ )))
+ }
+}
+
+impl<'de> Visitor<'de> for RootRowVisitor<'_> {
+ type Value = DecodeResult<GenericRow<'static>>;
+
+ fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str("a JSON row object")
+ }
+
+ fn visit_unit<E>(self) -> Result<Self::Value, E> {
+ Ok(self.type_error("null"))
+ }
+
+ fn visit_bool<E>(self, _value: bool) -> Result<Self::Value, E> {
+ Ok(self.type_error("a boolean"))
+ }
+
+ fn visit_i64<E>(self, _value: i64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_u64<E>(self, _value: u64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_f64<E>(self, _value: f64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_str<E>(self, _value: &str) -> Result<Self::Value, E> {
+ Ok(self.type_error("a string"))
+ }
+
+ fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
+ where
+ A: SeqAccess<'de>,
+ {
+ drain_sequence(&mut sequence)?;
+ Ok(self.type_error("an array"))
+ }
+
+ fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>
+ where
+ A: MapAccess<'de>,
+ {
+ decode_row_map(
+ map,
+ self.row_type,
+ RowContext::Root {
+ label: self.label,
+ column_indexes: self.column_indexes,
+ required: self.required,
+ },
+ )
+ }
+}
+
+enum RowContext<'a> {
+ Root {
+ label: &'a str,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+ },
+ Nested {
+ path: ValuePath<'a>,
+ },
+}
+
+impl<'a> RowContext<'a> {
+ fn field_index(&self, row_type: &RowType, name: &str) -> Option<usize> {
+ match self {
+ Self::Root { column_indexes, .. } => column_indexes.get(name).copied(),
+ Self::Nested { .. } => row_type.get_field_index(name),
+ }
+ }
+
+ fn field_path(&self, name: &str) -> ValuePath<'a> {
+ match self {
+ Self::Root { label, .. } => ValuePath::column(label, name),
+ Self::Nested { path } => path.nested_field(name),
+ }
+ }
+
+ fn required(&self, index: usize, data_type: &DataType) -> bool {
+ match self {
+ Self::Root { required, .. } => required[index],
+ Self::Nested { .. } => !data_type.is_nullable(),
+ }
+ }
+
+ fn duplicate(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{label}: duplicate column `{name}`"),
+ )),
+ Self::Nested { path } => path.invalid(format!("has a duplicate field `{name}`")),
+ }
+ }
+
+ fn unknown(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::schema_mismatch(
+ GatewayError::invalid_argument(format!("{label}: unknown column `{name}`")),
+ ),
+ Self::Nested { path } => path.schema_mismatch(format!("has an unknown field `{name}`")),
+ }
+ }
+
+ fn missing(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => {
+ RowDecodeError::schema_mismatch(GatewayError::invalid_argument(format!(
+ "{label}: column `{name}` is required and was not provided"
+ )))
+ }
+ Self::Nested { path } => path
+ .nested_field(name)
+ .schema_mismatch("is required and was not provided"),
+ }
+ }
+
+ fn required_null(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{label}: column `{name}` is required and must not be null"),
+ )),
+ Self::Nested { path } => path.nested_field(name).invalid("must not be null"),
+ }
+ }
+}
+
+fn decode_row_map<'de, A>(
+ mut map: A,
+ row_type: &RowType,
+ context: RowContext<'_>,
+) -> Result<DecodeResult<GenericRow<'static>>, A::Error>
+where
+ A: MapAccess<'de>,
+{
+ let mut seen = HashSet::with_capacity(map.size_hint().unwrap_or(0));
+ let mut duplicate = None;
+ let mut unknown = None;
+ let mut provided = std::iter::repeat_with(|| None)
+ .take(row_type.fields().len())
+ .collect::<Vec<Option<DecodeResult<Datum<'static>>>>>();
+
+ while let Some(name) = map.next_key::<String>()? {
+ if !seen.insert(name.clone()) {
+ map.next_value::<IgnoredAny>()?;
+ duplicate.get_or_insert(name);
+ continue;
+ }
+ let Some(index) = context.field_index(row_type, &name) else {
+ map.next_value::<IgnoredAny>()?;
+ unknown.get_or_insert(name);
+ continue;
+ };
+ let field = &row_type.fields()[index];
+ let value = map.next_value_seed(ValueSeed {
+ path: context.field_path(&field.name),
+ data_type: &field.data_type,
+ })?;
+ provided[index] = Some(value);
+ }
+
+ if let Some(name) = duplicate {
+ return Ok(Err(context.duplicate(&name)));
+ }
+ if let Some(name) = unknown {
+ return Ok(Err(context.unknown(&name)));
+ }
+
+ let mut row = GenericRow::new(row_type.fields().len());
+ for (index, field) in row_type.fields().iter().enumerate() {
+ let datum = match provided[index].take() {
+ Some(Ok(Datum::Null)) if context.required(index, &field.data_type) => {
+ return Ok(Err(context.required_null(&field.name)));
+ }
+ Some(Ok(datum)) => datum,
+ Some(Err(error)) => return Ok(Err(error)),
+ None if context.required(index, &field.data_type) => {
+ return Ok(Err(context.missing(&field.name)));
+ }
+ None => Datum::Null,
+ };
+ row.set_field(index, datum);
+ }
+ Ok(Ok(row))
+}
+
+struct ValueSeed<'a> {
+ path: ValuePath<'a>,
+ data_type: &'a DataType,
+}
+
+impl<'de> DeserializeSeed<'de> for ValueSeed<'_> {
+ type Value = DecodeResult<Datum<'static>>;
+
+ fn deserialize<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
+ where
+ D: de::Deserializer<'de>,
+ {
+ if is_complex(self.data_type) {
+ deserializer.deserialize_any(ComplexValueVisitor {
+ path: self.path,
+ data_type: self.data_type,
+ })
+ } else {
+ let raw = <&RawValue>::deserialize(deserializer)?;
+ let value = parse_scalar_value(raw).map_err(de::Error::custom)?;
+ Ok(decode_scalar(&self.path, self.data_type, &value))
+ }
+ }
+}
+
+struct ComplexValueVisitor<'a> {
+ path: ValuePath<'a>,
+ data_type: &'a DataType,
+}
+
+impl ComplexValueVisitor<'_> {
+ fn type_error(&self, actual: &str) -> DecodeResult<Datum<'static>> {
+ Err(self
+ .path
+ .type_error(complex_expected(self.data_type), actual))
+ }
+
+ fn null(&self) -> DecodeResult<Datum<'static>> {
+ if self.data_type.is_nullable() {
+ Ok(Datum::Null)
+ } else {
+ Err(self.path.invalid("must not be null"))
+ }
+ }
+}
+
+impl<'de> Visitor<'de> for ComplexValueVisitor<'_> {
+ type Value = DecodeResult<Datum<'static>>;
+
+ fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str(complex_expected(self.data_type))
+ }
+
+ fn visit_unit<E>(self) -> Result<Self::Value, E> {
+ Ok(self.null())
+ }
+
+ fn visit_bool<E>(self, _value: bool) -> Result<Self::Value, E> {
+ Ok(self.type_error("a boolean"))
+ }
+
+ fn visit_i64<E>(self, _value: i64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_u64<E>(self, _value: u64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_f64<E>(self, _value: f64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_str<E>(self, _value: &str) -> Result<Self::Value, E> {
+ Ok(self.type_error("a string"))
+ }
+
+ fn visit_seq<A>(self, sequence: A) -> Result<Self::Value, A::Error>
+ where
+ A: SeqAccess<'de>,
+ {
+ match self.data_type {
+ DataType::Array(data_type) => {
+ decode_array_sequence(sequence, &self.path, data_type.get_element_type())
+ }
+ DataType::Map(data_type) => decode_map_sequence(
+ sequence,
+ &self.path,
+ data_type.key_type(),
+ data_type.value_type(),
+ ),
+ DataType::Row(_) => {
+ let mut sequence = sequence;
+ drain_sequence(&mut sequence)?;
+ Ok(self.type_error("an array"))
+ }
+ _ => unreachable!("complex visitor receives only ARRAY, MAP, or ROW"),
+ }
+ }
+
+ fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>
+ where
+ A: MapAccess<'de>,
+ {
+ match self.data_type {
+ DataType::Row(row_type) => {
+ let row = decode_row_map(map, row_type, RowContext::Nested { path: self.path })?;
+ Ok(row.map(|row| Datum::Row(Box::new(row))))
+ }
+ DataType::Array(_) | DataType::Map(_) => {
+ let mut map = map;
+ drain_map(&mut map)?;
+ Ok(self.type_error("an object"))
+ }
+ _ => unreachable!("complex visitor receives only ARRAY, MAP, or ROW"),
+ }
+ }
+}
+
+fn complex_expected(data_type: &DataType) -> &'static str {
+ match data_type {
+ DataType::Array(_) => "ARRAY (a JSON array)",
+ DataType::Map(_) => "MAP (an array of {key, value} objects)",
+ DataType::Row(_) => "ROW (a JSON object of its declared fields)",
+ _ => unreachable!("only complex data types have a container expectation"),
+ }
+}
+
+fn drain_sequence<'de, A>(sequence: &mut A) -> Result<(), A::Error>
+where
+ A: SeqAccess<'de>,
+{
+ while sequence.next_element::<IgnoredAny>()?.is_some() {}
+ Ok(())
+}
+
+fn drain_map<'de, A>(map: &mut A) -> Result<(), A::Error>
+where
+ A: MapAccess<'de>,
+{
+ while map.next_entry::<IgnoredAny, IgnoredAny>()?.is_some() {}
+ Ok(())
+}
+
+fn decode_scalar(
+ path: &ValuePath<'_>,
+ data_type: &DataType,
+ value: &ScalarValue<'_>,
+) -> DecodeResult<Datum<'static>> {
+ if matches!(value, ScalarValue::Null) {
+ return if data_type.is_nullable() {
+ Ok(Datum::Null)
+ } else {
+ Err(path.invalid("must not be null"))
+ };
+ }
+ match data_type {
+ DataType::Boolean(_) => match value {
+ ScalarValue::Boolean(parsed) => Ok(Datum::Bool(*parsed)),
+ _ => Err(path.type_error("BOOLEAN (a JSON boolean)", value.kind())),
+ },
+ DataType::TinyInt(_) => {
+ decode_integer(path, value, "TINYINT", i8::MIN as i64, i8::MAX as i64)
+ .map(|parsed| Datum::Int8(parsed as i8))
+ }
+ DataType::SmallInt(_) => {
+ decode_integer(path, value, "SMALLINT", i16::MIN as i64, i16::MAX as i64)
+ .map(|parsed| Datum::Int16(parsed as i16))
+ }
+ DataType::Int(_) => decode_integer(path, value, "INT", i32::MIN as i64, i32::MAX as i64)
+ .map(|parsed| Datum::Int32(parsed as i32)),
+ DataType::BigInt(_) => decode_bigint(path, value).map(Datum::Int64),
+ DataType::Float(_) => decode_float32(path, value),
+ DataType::Double(_) => decode_float(path, value, "DOUBLE").map(Datum::from),
+ DataType::Char(_) | DataType::String(_) => match value {
+ ScalarValue::String(text) => Ok(Datum::String(Cow::Owned(text.clone()))),
+ _ => Err(path.type_error("STRING (a JSON string)", value.kind())),
+ },
+ DataType::Decimal(data_type) => {
+ decode_decimal(path, value, data_type.precision(), data_type.scale())
+ }
+ DataType::Bytes(_) => decode_binary(path, value, None),
+ DataType::Binary(data_type) => decode_binary(path, value, Some(data_type.length())),
+ DataType::Date(_) => decode_date(path, value),
+ DataType::Time(data_type) => decode_time(path, value, data_type.precision()),
+ DataType::Timestamp(data_type) => {
+ decode_timestamp(path, value, data_type.precision(), false)
+ }
+ DataType::TimestampLTz(data_type) => {
+ decode_timestamp(path, value, data_type.precision(), true)
+ }
+ DataType::Array(_) | DataType::Map(_) | DataType::Row(_) => {
+ unreachable!("container values use the complex visitor")
+ }
+ }
+}
+fn decode_integer(
+ path: &ValuePath<'_>,
+ value: &ScalarValue<'_>,
+ type_name: &str,
+ min: i64,
+ max: i64,
+) -> Result<i64, RowDecodeError> {
+ let expected = format!("{type_name} (an integer in [{min}, {max}])");
Review Comment:
Then "42" is accepted into a BIGINT but rejected into an INT, and "1.5" is rejected into a DOUBLE while "NaN" works.
Should the same JSON value depend on the column width?
##########
fluss-gateway/src/protocol/rest/codec/decoder.rs:
##########
@@ -0,0 +1,1927 @@
+// 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.
+
+//! Schema-aware streaming conversion from one raw FIP-49 JSON row to [`GenericRow`].
+//!
+//! Serde owns JSON syntax and container traversal. Schema-matching ARRAY, MAP, and ROW values
+//! recurse only along the validated [`RowType`], bounded by [`MAX_TYPE_NESTING`]. Scalar and
+//! ignored subtrees use serde_json's iterative [`RawValue`] and [`IgnoredAny`] scan path, so JSON
+//! depth does not grow this decoder's call stack. No container subtree is reparsed, keeping work
+//! linear in row bytes apart from local scalar decoding; the HTTP layer separately bounds body
+//! size. Recheck the iterative-scan assumption when upgrading serde_json.
+
+use crate::error::GatewayError;
+use crate::protocol::rest::codec::temporal::{parse_date, parse_time, parse_timestamp};
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64;
+use fluss::metadata::{DataType, RowType};
+use fluss::row::{
+ Date, Datum, Decimal, FlussArrayWriter, FlussMapWriter, GenericRow, Time, TimestampLtz,
+ TimestampNtz,
+};
+use serde::Deserialize;
+use serde::de::{self, DeserializeSeed, IgnoredAny, MapAccess, SeqAccess, Visitor};
+use serde_json::value::RawValue;
+use std::borrow::Cow;
+use std::collections::{HashMap, HashSet};
+use std::fmt;
+use std::sync::Arc;
+
+/// Bounds schema-recursive validation and decoding; it is not a JSON body-depth limit.
+const MAX_TYPE_NESTING: usize = 64;
+const NANOS_PER_MILLI: i64 = 1_000_000;
+const MILLIS_PER_SECOND: i64 = 1_000;
+const MILLIS_PER_DAY: i64 = 86_400_000;
+
+/// Whether a row supplies a complete mutation or only selected fields.
+#[derive(Debug, Clone, Copy)]
+#[allow(dead_code)]
+pub(crate) enum RowShape<'a> {
+ /// Every non-nullable column must be present. Missing nullable columns become null.
+ Complete,
+ /// Only the named columns must be present and non-null. Other missing columns become null.
+ Sparse(&'a [String]),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum RowDecodeErrorKind {
+ SchemaMismatch,
+ InvalidValue,
+}
+
+/// One row decode failure plus whether refreshed metadata could plausibly resolve it.
+#[derive(Debug)]
+pub(crate) struct RowDecodeError {
+ kind: RowDecodeErrorKind,
+ error: GatewayError,
+}
+
+#[allow(dead_code)]
+impl RowDecodeError {
+ fn schema_mismatch(error: GatewayError) -> Self {
+ Self {
+ kind: RowDecodeErrorKind::SchemaMismatch,
+ error,
+ }
+ }
+
+ fn invalid(error: GatewayError) -> Self {
+ Self {
+ kind: RowDecodeErrorKind::InvalidValue,
+ error,
+ }
+ }
+
+ /// Returns true only for failures that a newer table schema may resolve.
+ pub(crate) fn is_schema_mismatch(&self) -> bool {
+ self.kind == RowDecodeErrorKind::SchemaMismatch
+ }
+
+ /// Returns the client-safe error message.
+ pub(crate) fn message(&self) -> &str {
+ self.error.message()
+ }
+
+ /// Converts this failure into the standard Gateway error envelope input.
+ pub(crate) fn into_gateway_error(self) -> GatewayError {
+ self.error
+ }
+}
+
+impl From<GatewayError> for RowDecodeError {
+ fn from(error: GatewayError) -> Self {
+ Self::invalid(error)
+ }
+}
+
+struct SchemaDecoderInner {
+ row_type: RowType,
+ column_indexes: HashMap<String, usize>,
+}
+
+/// Immutable reusable decoder compiled from one Fluss row type.
+#[derive(Clone)]
+#[allow(dead_code)]
+pub(crate) struct SchemaDecoder {
+ inner: Arc<SchemaDecoderInner>,
+}
+
+impl fmt::Debug for SchemaDecoder {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter
+ .debug_struct("SchemaDecoder")
+ .field("row_type", &self.inner.row_type)
+ .finish()
+ }
+}
+
+#[allow(dead_code)]
+impl SchemaDecoder {
+ /// Builds a decoder for a complete table row type.
+ pub(crate) fn new(row_type: RowType) -> Result<Self, GatewayError> {
+ validate_row_type(&row_type, 0)?;
+ if row_type.fields().is_empty() {
+ return Err(GatewayError::internal(
+ "cannot build a row decoder for an empty table schema",
+ ));
+ }
+ let column_indexes = row_type
+ .fields()
+ .iter()
+ .enumerate()
+ .map(|(index, field)| (field.name.clone(), index))
+ .collect();
+ Ok(Self {
+ inner: Arc::new(SchemaDecoderInner {
+ row_type,
+ column_indexes,
+ }),
+ })
+ }
+
+ /// Returns the exact row type this decoder was built from.
+ pub(crate) fn row_type(&self) -> &RowType {
+ &self.inner.row_type
+ }
+
+ /// Decodes one raw JSON object into complete schema order.
+ pub(crate) fn decode_row(
+ &self,
+ label: &str,
+ row_json: &[u8],
+ shape: RowShape<'_>,
+ ) -> Result<GenericRow<'static>, RowDecodeError> {
+ let required = self.required_columns(label, shape)?;
+ let seed = RootRowSeed {
+ label,
+ row_type: &self.inner.row_type,
+ column_indexes: &self.inner.column_indexes,
+ required: &required,
+ };
+ let mut deserializer = serde_json::Deserializer::from_slice(row_json);
+ let decoded = seed
+ .deserialize(&mut deserializer)
+ .map_err(|error| invalid_json_error(label, error))?;
+ deserializer
+ .end()
+ .map_err(|error| invalid_json_error(label, error))?;
+ decoded
+ }
+
+ fn required_columns(
+ &self,
+ label: &str,
+ shape: RowShape<'_>,
+ ) -> Result<Vec<bool>, RowDecodeError> {
+ match shape {
+ RowShape::Complete => Ok(self
+ .inner
+ .row_type
+ .fields()
+ .iter()
+ .map(|field| !field.data_type.is_nullable())
+ .collect()),
+ RowShape::Sparse(required_columns) => {
+ let mut required = vec![false; self.inner.row_type.fields().len()];
+ for column in required_columns {
+ let Some(index) = self.inner.column_indexes.get(column) else {
+ return Err(RowDecodeError::schema_mismatch(
+ GatewayError::invalid_argument(format!(
+ "{label}: required column `{column}` is not part of the table schema"
+ )),
+ ));
+ };
+ required[*index] = true;
+ }
+ Ok(required)
+ }
+ }
+ }
+}
+
+fn invalid_json_error(label: &str, error: serde_json::Error) -> RowDecodeError {
+ RowDecodeError::invalid(GatewayError::invalid_argument(format!(
+ "{label}: invalid JSON row: {error}"
+ )))
+}
+
+fn validate_row_type(row_type: &RowType, depth: usize) -> Result<(), GatewayError> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::internal(format!(
+ "row type nesting exceeds {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ let mut names = HashSet::with_capacity(row_type.fields().len());
+ for field in row_type.fields() {
+ if field.name.is_empty() {
+ return Err(GatewayError::internal(
+ "row type contains an empty field name",
+ ));
+ }
+ if !names.insert(field.name.as_str()) {
+ return Err(GatewayError::internal(format!(
+ "row type contains duplicate field `{}`",
+ field.name
+ )));
+ }
+ validate_data_type(&field.data_type, depth + 1)?;
+ }
+ Ok(())
+}
+
+fn validate_data_type(data_type: &DataType, depth: usize) -> Result<(), GatewayError> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::internal(format!(
+ "row type nesting exceeds {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ match data_type {
+ DataType::Char(data_type) if data_type.length() == 0 => {
+ Err(GatewayError::internal("CHAR length must be at least one"))
+ }
+ DataType::Binary(data_type) if data_type.length() == 0 => {
+ Err(GatewayError::internal("BINARY length must be at least one"))
+ }
+ DataType::Array(data_type) => validate_data_type(data_type.get_element_type(), depth + 1),
+ DataType::Map(data_type) => {
+ if data_type.key_type().is_nullable() {
+ return Err(GatewayError::internal("MAP key type must not be nullable"));
+ }
+ if is_complex(data_type.key_type()) {
+ return Err(GatewayError::invalid_argument(
+ "FIP-49 JSON does not support ARRAY, MAP, or ROW values as MAP keys",
+ ));
+ }
+ validate_data_type(data_type.key_type(), depth + 1)?;
+ validate_data_type(data_type.value_type(), depth + 1)
+ }
+ DataType::Row(row_type) => validate_row_type(row_type, depth + 1),
+ _ => Ok(()),
+ }
+}
+
+fn is_complex(data_type: &DataType) -> bool {
+ matches!(
+ data_type,
+ DataType::Array(_) | DataType::Map(_) | DataType::Row(_)
+ )
+}
+
+#[derive(Clone)]
+struct ValuePath<'a> {
+ label: &'a str,
+ path: String,
+}
+
+impl<'a> ValuePath<'a> {
+ fn column(label: &'a str, column: &str) -> Self {
+ Self {
+ label,
+ path: column.to_string(),
+ }
+ }
+
+ fn element(&self, index: usize) -> Self {
+ self.nested(format!("{}[{index}]", self.path))
+ }
+
+ fn nested_field(&self, name: &str) -> Self {
+ self.nested(format!("{}.{name}", self.path))
+ }
+
+ fn map_part(&self, index: usize, part: &str) -> Self {
+ self.nested(format!("{}[{index}].{part}", self.path))
+ }
+
+ fn nested(&self, path: String) -> Self {
+ Self {
+ label: self.label,
+ path,
+ }
+ }
+
+ fn invalid(&self, reason: impl fmt::Display) -> RowDecodeError {
+ RowDecodeError::invalid(GatewayError::invalid_argument(format!(
+ "{}: column `{}` {reason}",
+ self.label, self.path
+ )))
+ }
+
+ fn schema_mismatch(&self, reason: impl fmt::Display) -> RowDecodeError {
+ RowDecodeError::schema_mismatch(GatewayError::invalid_argument(format!(
+ "{}: column `{}` {reason}",
+ self.label, self.path
+ )))
+ }
+
+ fn type_error(&self, expected: &str, actual: &str) -> RowDecodeError {
+ self.invalid(format!("expects {expected}, got {actual}"))
+ }
+}
+
+type DecodeResult<T> = Result<T, RowDecodeError>;
+
+enum ScalarValue<'a> {
+ Null,
+ Boolean(bool),
+ ExactNumber(&'a str),
+ String(String),
+ Array,
+ Object,
+}
+
+impl ScalarValue<'_> {
+ fn kind(&self) -> &'static str {
+ match self {
+ Self::Null => "null",
+ Self::Boolean(_) => "a boolean",
+ Self::ExactNumber(_) => "a number",
+ Self::String(_) => "a string",
+ Self::Array => "an array",
+ Self::Object => "an object",
+ }
+ }
+}
+
+fn parse_scalar_value(raw: &RawValue) -> Result<ScalarValue<'_>, serde_json::Error> {
+ let text = raw.get().trim();
+ match text.as_bytes().first().copied() {
+ Some(b'n') => Ok(ScalarValue::Null),
+ Some(b't') => Ok(ScalarValue::Boolean(true)),
+ Some(b'f') => Ok(ScalarValue::Boolean(false)),
+ Some(b'"') => serde_json::from_str(text).map(ScalarValue::String),
+ Some(b'-' | b'0'..=b'9') => Ok(ScalarValue::ExactNumber(text)),
+ Some(b'[') => Ok(ScalarValue::Array),
+ Some(b'{') => Ok(ScalarValue::Object),
+ _ => unreachable!("RawValue always contains one complete JSON value"),
+ }
+}
+
+struct RootRowSeed<'a> {
+ label: &'a str,
+ row_type: &'a RowType,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+}
+
+impl<'de> DeserializeSeed<'de> for RootRowSeed<'_> {
+ type Value = DecodeResult<GenericRow<'static>>;
+
+ fn deserialize<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
+ where
+ D: de::Deserializer<'de>,
+ {
+ deserializer.deserialize_any(RootRowVisitor {
+ label: self.label,
+ row_type: self.row_type,
+ column_indexes: self.column_indexes,
+ required: self.required,
+ })
+ }
+}
+
+struct RootRowVisitor<'a> {
+ label: &'a str,
+ row_type: &'a RowType,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+}
+
+impl RootRowVisitor<'_> {
+ fn type_error(&self, actual: &str) -> DecodeResult<GenericRow<'static>> {
+ Err(RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{}: row must be a JSON object, got {actual}", self.label),
+ )))
+ }
+}
+
+impl<'de> Visitor<'de> for RootRowVisitor<'_> {
+ type Value = DecodeResult<GenericRow<'static>>;
+
+ fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str("a JSON row object")
+ }
+
+ fn visit_unit<E>(self) -> Result<Self::Value, E> {
+ Ok(self.type_error("null"))
+ }
+
+ fn visit_bool<E>(self, _value: bool) -> Result<Self::Value, E> {
+ Ok(self.type_error("a boolean"))
+ }
+
+ fn visit_i64<E>(self, _value: i64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_u64<E>(self, _value: u64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_f64<E>(self, _value: f64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_str<E>(self, _value: &str) -> Result<Self::Value, E> {
+ Ok(self.type_error("a string"))
+ }
+
+ fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
+ where
+ A: SeqAccess<'de>,
+ {
+ drain_sequence(&mut sequence)?;
+ Ok(self.type_error("an array"))
+ }
+
+ fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>
+ where
+ A: MapAccess<'de>,
+ {
+ decode_row_map(
+ map,
+ self.row_type,
+ RowContext::Root {
+ label: self.label,
+ column_indexes: self.column_indexes,
+ required: self.required,
+ },
+ )
+ }
+}
+
+enum RowContext<'a> {
+ Root {
+ label: &'a str,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+ },
+ Nested {
+ path: ValuePath<'a>,
+ },
+}
+
+impl<'a> RowContext<'a> {
+ fn field_index(&self, row_type: &RowType, name: &str) -> Option<usize> {
+ match self {
+ Self::Root { column_indexes, .. } => column_indexes.get(name).copied(),
+ Self::Nested { .. } => row_type.get_field_index(name),
+ }
+ }
+
+ fn field_path(&self, name: &str) -> ValuePath<'a> {
+ match self {
+ Self::Root { label, .. } => ValuePath::column(label, name),
+ Self::Nested { path } => path.nested_field(name),
+ }
+ }
+
+ fn required(&self, index: usize, data_type: &DataType) -> bool {
+ match self {
+ Self::Root { required, .. } => required[index],
+ Self::Nested { .. } => !data_type.is_nullable(),
+ }
+ }
+
+ fn duplicate(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{label}: duplicate column `{name}`"),
+ )),
+ Self::Nested { path } => path.invalid(format!("has a duplicate field `{name}`")),
+ }
+ }
+
+ fn unknown(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::schema_mismatch(
+ GatewayError::invalid_argument(format!("{label}: unknown column `{name}`")),
+ ),
+ Self::Nested { path } => path.schema_mismatch(format!("has an unknown field `{name}`")),
+ }
+ }
+
+ fn missing(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => {
+ RowDecodeError::schema_mismatch(GatewayError::invalid_argument(format!(
+ "{label}: column `{name}` is required and was not provided"
+ )))
+ }
+ Self::Nested { path } => path
+ .nested_field(name)
+ .schema_mismatch("is required and was not provided"),
+ }
+ }
+
+ fn required_null(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{label}: column `{name}` is required and must not be null"),
+ )),
+ Self::Nested { path } => path.nested_field(name).invalid("must not be null"),
+ }
+ }
+}
+
+fn decode_row_map<'de, A>(
+ mut map: A,
+ row_type: &RowType,
+ context: RowContext<'_>,
+) -> Result<DecodeResult<GenericRow<'static>>, A::Error>
+where
+ A: MapAccess<'de>,
+{
+ let mut seen = HashSet::with_capacity(map.size_hint().unwrap_or(0));
+ let mut duplicate = None;
+ let mut unknown = None;
+ let mut provided = std::iter::repeat_with(|| None)
+ .take(row_type.fields().len())
+ .collect::<Vec<Option<DecodeResult<Datum<'static>>>>>();
+
+ while let Some(name) = map.next_key::<String>()? {
+ if !seen.insert(name.clone()) {
+ map.next_value::<IgnoredAny>()?;
+ duplicate.get_or_insert(name);
+ continue;
+ }
+ let Some(index) = context.field_index(row_type, &name) else {
+ map.next_value::<IgnoredAny>()?;
+ unknown.get_or_insert(name);
+ continue;
+ };
+ let field = &row_type.fields()[index];
+ let value = map.next_value_seed(ValueSeed {
+ path: context.field_path(&field.name),
+ data_type: &field.data_type,
+ })?;
+ provided[index] = Some(value);
+ }
+
+ if let Some(name) = duplicate {
+ return Ok(Err(context.duplicate(&name)));
+ }
+ if let Some(name) = unknown {
+ return Ok(Err(context.unknown(&name)));
+ }
+
+ let mut row = GenericRow::new(row_type.fields().len());
+ for (index, field) in row_type.fields().iter().enumerate() {
+ let datum = match provided[index].take() {
+ Some(Ok(Datum::Null)) if context.required(index, &field.data_type) => {
+ return Ok(Err(context.required_null(&field.name)));
+ }
+ Some(Ok(datum)) => datum,
+ Some(Err(error)) => return Ok(Err(error)),
+ None if context.required(index, &field.data_type) => {
+ return Ok(Err(context.missing(&field.name)));
+ }
+ None => Datum::Null,
+ };
+ row.set_field(index, datum);
+ }
+ Ok(Ok(row))
+}
+
+struct ValueSeed<'a> {
+ path: ValuePath<'a>,
+ data_type: &'a DataType,
+}
+
+impl<'de> DeserializeSeed<'de> for ValueSeed<'_> {
+ type Value = DecodeResult<Datum<'static>>;
+
+ fn deserialize<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
+ where
+ D: de::Deserializer<'de>,
+ {
+ if is_complex(self.data_type) {
+ deserializer.deserialize_any(ComplexValueVisitor {
+ path: self.path,
+ data_type: self.data_type,
+ })
+ } else {
+ let raw = <&RawValue>::deserialize(deserializer)?;
+ let value = parse_scalar_value(raw).map_err(de::Error::custom)?;
+ Ok(decode_scalar(&self.path, self.data_type, &value))
+ }
+ }
+}
+
+struct ComplexValueVisitor<'a> {
+ path: ValuePath<'a>,
+ data_type: &'a DataType,
+}
+
+impl ComplexValueVisitor<'_> {
+ fn type_error(&self, actual: &str) -> DecodeResult<Datum<'static>> {
+ Err(self
+ .path
+ .type_error(complex_expected(self.data_type), actual))
+ }
+
+ fn null(&self) -> DecodeResult<Datum<'static>> {
+ if self.data_type.is_nullable() {
+ Ok(Datum::Null)
+ } else {
+ Err(self.path.invalid("must not be null"))
+ }
+ }
+}
+
+impl<'de> Visitor<'de> for ComplexValueVisitor<'_> {
+ type Value = DecodeResult<Datum<'static>>;
+
+ fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str(complex_expected(self.data_type))
+ }
+
+ fn visit_unit<E>(self) -> Result<Self::Value, E> {
+ Ok(self.null())
+ }
+
+ fn visit_bool<E>(self, _value: bool) -> Result<Self::Value, E> {
+ Ok(self.type_error("a boolean"))
+ }
+
+ fn visit_i64<E>(self, _value: i64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_u64<E>(self, _value: u64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_f64<E>(self, _value: f64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_str<E>(self, _value: &str) -> Result<Self::Value, E> {
+ Ok(self.type_error("a string"))
+ }
+
+ fn visit_seq<A>(self, sequence: A) -> Result<Self::Value, A::Error>
+ where
+ A: SeqAccess<'de>,
+ {
+ match self.data_type {
+ DataType::Array(data_type) => {
+ decode_array_sequence(sequence, &self.path, data_type.get_element_type())
+ }
+ DataType::Map(data_type) => decode_map_sequence(
+ sequence,
+ &self.path,
+ data_type.key_type(),
+ data_type.value_type(),
+ ),
+ DataType::Row(_) => {
+ let mut sequence = sequence;
+ drain_sequence(&mut sequence)?;
+ Ok(self.type_error("an array"))
+ }
+ _ => unreachable!("complex visitor receives only ARRAY, MAP, or ROW"),
+ }
+ }
+
+ fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>
+ where
+ A: MapAccess<'de>,
+ {
+ match self.data_type {
+ DataType::Row(row_type) => {
+ let row = decode_row_map(map, row_type, RowContext::Nested { path: self.path })?;
+ Ok(row.map(|row| Datum::Row(Box::new(row))))
+ }
+ DataType::Array(_) | DataType::Map(_) => {
+ let mut map = map;
+ drain_map(&mut map)?;
+ Ok(self.type_error("an object"))
+ }
+ _ => unreachable!("complex visitor receives only ARRAY, MAP, or ROW"),
+ }
+ }
+}
+
+fn complex_expected(data_type: &DataType) -> &'static str {
+ match data_type {
+ DataType::Array(_) => "ARRAY (a JSON array)",
+ DataType::Map(_) => "MAP (an array of {key, value} objects)",
+ DataType::Row(_) => "ROW (a JSON object of its declared fields)",
+ _ => unreachable!("only complex data types have a container expectation"),
+ }
+}
+
+fn drain_sequence<'de, A>(sequence: &mut A) -> Result<(), A::Error>
+where
+ A: SeqAccess<'de>,
+{
+ while sequence.next_element::<IgnoredAny>()?.is_some() {}
+ Ok(())
+}
+
+fn drain_map<'de, A>(map: &mut A) -> Result<(), A::Error>
+where
+ A: MapAccess<'de>,
+{
+ while map.next_entry::<IgnoredAny, IgnoredAny>()?.is_some() {}
+ Ok(())
+}
+
+fn decode_scalar(
+ path: &ValuePath<'_>,
+ data_type: &DataType,
+ value: &ScalarValue<'_>,
+) -> DecodeResult<Datum<'static>> {
+ if matches!(value, ScalarValue::Null) {
+ return if data_type.is_nullable() {
+ Ok(Datum::Null)
+ } else {
+ Err(path.invalid("must not be null"))
+ };
+ }
+ match data_type {
+ DataType::Boolean(_) => match value {
+ ScalarValue::Boolean(parsed) => Ok(Datum::Bool(*parsed)),
+ _ => Err(path.type_error("BOOLEAN (a JSON boolean)", value.kind())),
+ },
+ DataType::TinyInt(_) => {
+ decode_integer(path, value, "TINYINT", i8::MIN as i64, i8::MAX as i64)
+ .map(|parsed| Datum::Int8(parsed as i8))
+ }
+ DataType::SmallInt(_) => {
+ decode_integer(path, value, "SMALLINT", i16::MIN as i64, i16::MAX as i64)
+ .map(|parsed| Datum::Int16(parsed as i16))
+ }
+ DataType::Int(_) => decode_integer(path, value, "INT", i32::MIN as i64, i32::MAX as i64)
+ .map(|parsed| Datum::Int32(parsed as i32)),
+ DataType::BigInt(_) => decode_bigint(path, value).map(Datum::Int64),
+ DataType::Float(_) => decode_float32(path, value),
+ DataType::Double(_) => decode_float(path, value, "DOUBLE").map(Datum::from),
+ DataType::Char(_) | DataType::String(_) => match value {
Review Comment:
BINARY(2) rejects a 5-byte value on line 782, but CHAR(3) accepts "abcd" and passes it straight through. Should CHAR check its length too?
##########
fluss-gateway/src/protocol/rest/codec/decoder.rs:
##########
@@ -0,0 +1,1927 @@
+// 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.
+
+//! Schema-aware streaming conversion from one raw FIP-49 JSON row to [`GenericRow`].
+//!
+//! Serde owns JSON syntax and container traversal. Schema-matching ARRAY, MAP, and ROW values
+//! recurse only along the validated [`RowType`], bounded by [`MAX_TYPE_NESTING`]. Scalar and
+//! ignored subtrees use serde_json's iterative [`RawValue`] and [`IgnoredAny`] scan path, so JSON
+//! depth does not grow this decoder's call stack. No container subtree is reparsed, keeping work
+//! linear in row bytes apart from local scalar decoding; the HTTP layer separately bounds body
+//! size. Recheck the iterative-scan assumption when upgrading serde_json.
+
+use crate::error::GatewayError;
+use crate::protocol::rest::codec::temporal::{parse_date, parse_time, parse_timestamp};
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64;
+use fluss::metadata::{DataType, RowType};
+use fluss::row::{
+ Date, Datum, Decimal, FlussArrayWriter, FlussMapWriter, GenericRow, Time, TimestampLtz,
+ TimestampNtz,
+};
+use serde::Deserialize;
+use serde::de::{self, DeserializeSeed, IgnoredAny, MapAccess, SeqAccess, Visitor};
+use serde_json::value::RawValue;
+use std::borrow::Cow;
+use std::collections::{HashMap, HashSet};
+use std::fmt;
+use std::sync::Arc;
+
+/// Bounds schema-recursive validation and decoding; it is not a JSON body-depth limit.
+const MAX_TYPE_NESTING: usize = 64;
+const NANOS_PER_MILLI: i64 = 1_000_000;
+const MILLIS_PER_SECOND: i64 = 1_000;
+const MILLIS_PER_DAY: i64 = 86_400_000;
+
+/// Whether a row supplies a complete mutation or only selected fields.
+#[derive(Debug, Clone, Copy)]
+#[allow(dead_code)]
+pub(crate) enum RowShape<'a> {
+ /// Every non-nullable column must be present. Missing nullable columns become null.
+ Complete,
+ /// Only the named columns must be present and non-null. Other missing columns become null.
+ Sparse(&'a [String]),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum RowDecodeErrorKind {
+ SchemaMismatch,
+ InvalidValue,
+}
+
+/// One row decode failure plus whether refreshed metadata could plausibly resolve it.
+#[derive(Debug)]
+pub(crate) struct RowDecodeError {
+ kind: RowDecodeErrorKind,
+ error: GatewayError,
+}
+
+#[allow(dead_code)]
+impl RowDecodeError {
+ fn schema_mismatch(error: GatewayError) -> Self {
+ Self {
+ kind: RowDecodeErrorKind::SchemaMismatch,
+ error,
+ }
+ }
+
+ fn invalid(error: GatewayError) -> Self {
+ Self {
+ kind: RowDecodeErrorKind::InvalidValue,
+ error,
+ }
+ }
+
+ /// Returns true only for failures that a newer table schema may resolve.
+ pub(crate) fn is_schema_mismatch(&self) -> bool {
+ self.kind == RowDecodeErrorKind::SchemaMismatch
+ }
+
+ /// Returns the client-safe error message.
+ pub(crate) fn message(&self) -> &str {
+ self.error.message()
+ }
+
+ /// Converts this failure into the standard Gateway error envelope input.
+ pub(crate) fn into_gateway_error(self) -> GatewayError {
+ self.error
+ }
+}
+
+impl From<GatewayError> for RowDecodeError {
+ fn from(error: GatewayError) -> Self {
+ Self::invalid(error)
+ }
+}
+
+struct SchemaDecoderInner {
+ row_type: RowType,
+ column_indexes: HashMap<String, usize>,
+}
+
+/// Immutable reusable decoder compiled from one Fluss row type.
+#[derive(Clone)]
+#[allow(dead_code)]
+pub(crate) struct SchemaDecoder {
+ inner: Arc<SchemaDecoderInner>,
+}
+
+impl fmt::Debug for SchemaDecoder {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter
+ .debug_struct("SchemaDecoder")
+ .field("row_type", &self.inner.row_type)
+ .finish()
+ }
+}
+
+#[allow(dead_code)]
+impl SchemaDecoder {
+ /// Builds a decoder for a complete table row type.
+ pub(crate) fn new(row_type: RowType) -> Result<Self, GatewayError> {
+ validate_row_type(&row_type, 0)?;
+ if row_type.fields().is_empty() {
+ return Err(GatewayError::internal(
+ "cannot build a row decoder for an empty table schema",
+ ));
+ }
+ let column_indexes = row_type
+ .fields()
+ .iter()
+ .enumerate()
+ .map(|(index, field)| (field.name.clone(), index))
+ .collect();
+ Ok(Self {
+ inner: Arc::new(SchemaDecoderInner {
+ row_type,
+ column_indexes,
+ }),
+ })
+ }
+
+ /// Returns the exact row type this decoder was built from.
+ pub(crate) fn row_type(&self) -> &RowType {
+ &self.inner.row_type
+ }
+
+ /// Decodes one raw JSON object into complete schema order.
+ pub(crate) fn decode_row(
+ &self,
+ label: &str,
+ row_json: &[u8],
+ shape: RowShape<'_>,
+ ) -> Result<GenericRow<'static>, RowDecodeError> {
+ let required = self.required_columns(label, shape)?;
+ let seed = RootRowSeed {
+ label,
+ row_type: &self.inner.row_type,
+ column_indexes: &self.inner.column_indexes,
+ required: &required,
+ };
+ let mut deserializer = serde_json::Deserializer::from_slice(row_json);
+ let decoded = seed
+ .deserialize(&mut deserializer)
+ .map_err(|error| invalid_json_error(label, error))?;
+ deserializer
+ .end()
+ .map_err(|error| invalid_json_error(label, error))?;
+ decoded
+ }
+
+ fn required_columns(
+ &self,
+ label: &str,
+ shape: RowShape<'_>,
+ ) -> Result<Vec<bool>, RowDecodeError> {
+ match shape {
+ RowShape::Complete => Ok(self
+ .inner
+ .row_type
+ .fields()
+ .iter()
+ .map(|field| !field.data_type.is_nullable())
+ .collect()),
+ RowShape::Sparse(required_columns) => {
+ let mut required = vec![false; self.inner.row_type.fields().len()];
+ for column in required_columns {
+ let Some(index) = self.inner.column_indexes.get(column) else {
+ return Err(RowDecodeError::schema_mismatch(
+ GatewayError::invalid_argument(format!(
+ "{label}: required column `{column}` is not part of the table schema"
+ )),
+ ));
+ };
+ required[*index] = true;
+ }
+ Ok(required)
+ }
+ }
+ }
+}
+
+fn invalid_json_error(label: &str, error: serde_json::Error) -> RowDecodeError {
+ RowDecodeError::invalid(GatewayError::invalid_argument(format!(
+ "{label}: invalid JSON row: {error}"
+ )))
+}
+
+fn validate_row_type(row_type: &RowType, depth: usize) -> Result<(), GatewayError> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::internal(format!(
+ "row type nesting exceeds {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ let mut names = HashSet::with_capacity(row_type.fields().len());
+ for field in row_type.fields() {
+ if field.name.is_empty() {
+ return Err(GatewayError::internal(
+ "row type contains an empty field name",
+ ));
+ }
+ if !names.insert(field.name.as_str()) {
+ return Err(GatewayError::internal(format!(
+ "row type contains duplicate field `{}`",
+ field.name
+ )));
+ }
+ validate_data_type(&field.data_type, depth + 1)?;
+ }
+ Ok(())
+}
+
+fn validate_data_type(data_type: &DataType, depth: usize) -> Result<(), GatewayError> {
+ if depth > MAX_TYPE_NESTING {
+ return Err(GatewayError::internal(format!(
+ "row type nesting exceeds {MAX_TYPE_NESTING} levels"
+ )));
+ }
+ match data_type {
+ DataType::Char(data_type) if data_type.length() == 0 => {
+ Err(GatewayError::internal("CHAR length must be at least one"))
+ }
+ DataType::Binary(data_type) if data_type.length() == 0 => {
+ Err(GatewayError::internal("BINARY length must be at least one"))
+ }
+ DataType::Array(data_type) => validate_data_type(data_type.get_element_type(), depth + 1),
+ DataType::Map(data_type) => {
+ if data_type.key_type().is_nullable() {
+ return Err(GatewayError::internal("MAP key type must not be nullable"));
+ }
+ if is_complex(data_type.key_type()) {
+ return Err(GatewayError::invalid_argument(
+ "FIP-49 JSON does not support ARRAY, MAP, or ROW values as MAP keys",
+ ));
+ }
+ validate_data_type(data_type.key_type(), depth + 1)?;
+ validate_data_type(data_type.value_type(), depth + 1)
+ }
+ DataType::Row(row_type) => validate_row_type(row_type, depth + 1),
+ _ => Ok(()),
+ }
+}
+
+fn is_complex(data_type: &DataType) -> bool {
+ matches!(
+ data_type,
+ DataType::Array(_) | DataType::Map(_) | DataType::Row(_)
+ )
+}
+
+#[derive(Clone)]
+struct ValuePath<'a> {
+ label: &'a str,
+ path: String,
+}
+
+impl<'a> ValuePath<'a> {
+ fn column(label: &'a str, column: &str) -> Self {
+ Self {
+ label,
+ path: column.to_string(),
+ }
+ }
+
+ fn element(&self, index: usize) -> Self {
+ self.nested(format!("{}[{index}]", self.path))
+ }
+
+ fn nested_field(&self, name: &str) -> Self {
+ self.nested(format!("{}.{name}", self.path))
+ }
+
+ fn map_part(&self, index: usize, part: &str) -> Self {
+ self.nested(format!("{}[{index}].{part}", self.path))
+ }
+
+ fn nested(&self, path: String) -> Self {
+ Self {
+ label: self.label,
+ path,
+ }
+ }
+
+ fn invalid(&self, reason: impl fmt::Display) -> RowDecodeError {
+ RowDecodeError::invalid(GatewayError::invalid_argument(format!(
+ "{}: column `{}` {reason}",
+ self.label, self.path
+ )))
+ }
+
+ fn schema_mismatch(&self, reason: impl fmt::Display) -> RowDecodeError {
+ RowDecodeError::schema_mismatch(GatewayError::invalid_argument(format!(
+ "{}: column `{}` {reason}",
+ self.label, self.path
+ )))
+ }
+
+ fn type_error(&self, expected: &str, actual: &str) -> RowDecodeError {
+ self.invalid(format!("expects {expected}, got {actual}"))
+ }
+}
+
+type DecodeResult<T> = Result<T, RowDecodeError>;
+
+enum ScalarValue<'a> {
+ Null,
+ Boolean(bool),
+ ExactNumber(&'a str),
+ String(String),
+ Array,
+ Object,
+}
+
+impl ScalarValue<'_> {
+ fn kind(&self) -> &'static str {
+ match self {
+ Self::Null => "null",
+ Self::Boolean(_) => "a boolean",
+ Self::ExactNumber(_) => "a number",
+ Self::String(_) => "a string",
+ Self::Array => "an array",
+ Self::Object => "an object",
+ }
+ }
+}
+
+fn parse_scalar_value(raw: &RawValue) -> Result<ScalarValue<'_>, serde_json::Error> {
+ let text = raw.get().trim();
+ match text.as_bytes().first().copied() {
+ Some(b'n') => Ok(ScalarValue::Null),
+ Some(b't') => Ok(ScalarValue::Boolean(true)),
+ Some(b'f') => Ok(ScalarValue::Boolean(false)),
+ Some(b'"') => serde_json::from_str(text).map(ScalarValue::String),
+ Some(b'-' | b'0'..=b'9') => Ok(ScalarValue::ExactNumber(text)),
+ Some(b'[') => Ok(ScalarValue::Array),
+ Some(b'{') => Ok(ScalarValue::Object),
+ _ => unreachable!("RawValue always contains one complete JSON value"),
+ }
+}
+
+struct RootRowSeed<'a> {
+ label: &'a str,
+ row_type: &'a RowType,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+}
+
+impl<'de> DeserializeSeed<'de> for RootRowSeed<'_> {
+ type Value = DecodeResult<GenericRow<'static>>;
+
+ fn deserialize<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
+ where
+ D: de::Deserializer<'de>,
+ {
+ deserializer.deserialize_any(RootRowVisitor {
+ label: self.label,
+ row_type: self.row_type,
+ column_indexes: self.column_indexes,
+ required: self.required,
+ })
+ }
+}
+
+struct RootRowVisitor<'a> {
+ label: &'a str,
+ row_type: &'a RowType,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+}
+
+impl RootRowVisitor<'_> {
+ fn type_error(&self, actual: &str) -> DecodeResult<GenericRow<'static>> {
+ Err(RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{}: row must be a JSON object, got {actual}", self.label),
+ )))
+ }
+}
+
+impl<'de> Visitor<'de> for RootRowVisitor<'_> {
+ type Value = DecodeResult<GenericRow<'static>>;
+
+ fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str("a JSON row object")
+ }
+
+ fn visit_unit<E>(self) -> Result<Self::Value, E> {
+ Ok(self.type_error("null"))
+ }
+
+ fn visit_bool<E>(self, _value: bool) -> Result<Self::Value, E> {
+ Ok(self.type_error("a boolean"))
+ }
+
+ fn visit_i64<E>(self, _value: i64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_u64<E>(self, _value: u64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_f64<E>(self, _value: f64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_str<E>(self, _value: &str) -> Result<Self::Value, E> {
+ Ok(self.type_error("a string"))
+ }
+
+ fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
+ where
+ A: SeqAccess<'de>,
+ {
+ drain_sequence(&mut sequence)?;
+ Ok(self.type_error("an array"))
+ }
+
+ fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>
+ where
+ A: MapAccess<'de>,
+ {
+ decode_row_map(
+ map,
+ self.row_type,
+ RowContext::Root {
+ label: self.label,
+ column_indexes: self.column_indexes,
+ required: self.required,
+ },
+ )
+ }
+}
+
+enum RowContext<'a> {
+ Root {
+ label: &'a str,
+ column_indexes: &'a HashMap<String, usize>,
+ required: &'a [bool],
+ },
+ Nested {
+ path: ValuePath<'a>,
+ },
+}
+
+impl<'a> RowContext<'a> {
+ fn field_index(&self, row_type: &RowType, name: &str) -> Option<usize> {
+ match self {
+ Self::Root { column_indexes, .. } => column_indexes.get(name).copied(),
+ Self::Nested { .. } => row_type.get_field_index(name),
+ }
+ }
+
+ fn field_path(&self, name: &str) -> ValuePath<'a> {
+ match self {
+ Self::Root { label, .. } => ValuePath::column(label, name),
+ Self::Nested { path } => path.nested_field(name),
+ }
+ }
+
+ fn required(&self, index: usize, data_type: &DataType) -> bool {
+ match self {
+ Self::Root { required, .. } => required[index],
+ Self::Nested { .. } => !data_type.is_nullable(),
+ }
+ }
+
+ fn duplicate(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{label}: duplicate column `{name}`"),
+ )),
+ Self::Nested { path } => path.invalid(format!("has a duplicate field `{name}`")),
+ }
+ }
+
+ fn unknown(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::schema_mismatch(
+ GatewayError::invalid_argument(format!("{label}: unknown column `{name}`")),
+ ),
+ Self::Nested { path } => path.schema_mismatch(format!("has an unknown field `{name}`")),
+ }
+ }
+
+ fn missing(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => {
+ RowDecodeError::schema_mismatch(GatewayError::invalid_argument(format!(
+ "{label}: column `{name}` is required and was not provided"
+ )))
+ }
+ Self::Nested { path } => path
+ .nested_field(name)
+ .schema_mismatch("is required and was not provided"),
+ }
+ }
+
+ fn required_null(&self, name: &str) -> RowDecodeError {
+ match self {
+ Self::Root { label, .. } => RowDecodeError::invalid(GatewayError::invalid_argument(
+ format!("{label}: column `{name}` is required and must not be null"),
+ )),
+ Self::Nested { path } => path.nested_field(name).invalid("must not be null"),
+ }
+ }
+}
+
+fn decode_row_map<'de, A>(
+ mut map: A,
+ row_type: &RowType,
+ context: RowContext<'_>,
+) -> Result<DecodeResult<GenericRow<'static>>, A::Error>
+where
+ A: MapAccess<'de>,
+{
+ let mut seen = HashSet::with_capacity(map.size_hint().unwrap_or(0));
+ let mut duplicate = None;
+ let mut unknown = None;
+ let mut provided = std::iter::repeat_with(|| None)
+ .take(row_type.fields().len())
+ .collect::<Vec<Option<DecodeResult<Datum<'static>>>>>();
+
+ while let Some(name) = map.next_key::<String>()? {
+ if !seen.insert(name.clone()) {
+ map.next_value::<IgnoredAny>()?;
+ duplicate.get_or_insert(name);
+ continue;
+ }
+ let Some(index) = context.field_index(row_type, &name) else {
+ map.next_value::<IgnoredAny>()?;
+ unknown.get_or_insert(name);
+ continue;
+ };
+ let field = &row_type.fields()[index];
+ let value = map.next_value_seed(ValueSeed {
+ path: context.field_path(&field.name),
+ data_type: &field.data_type,
+ })?;
+ provided[index] = Some(value);
+ }
+
+ if let Some(name) = duplicate {
+ return Ok(Err(context.duplicate(&name)));
+ }
+ if let Some(name) = unknown {
+ return Ok(Err(context.unknown(&name)));
+ }
+
+ let mut row = GenericRow::new(row_type.fields().len());
+ for (index, field) in row_type.fields().iter().enumerate() {
+ let datum = match provided[index].take() {
+ Some(Ok(Datum::Null)) if context.required(index, &field.data_type) => {
+ return Ok(Err(context.required_null(&field.name)));
+ }
+ Some(Ok(datum)) => datum,
+ Some(Err(error)) => return Ok(Err(error)),
+ None if context.required(index, &field.data_type) => {
+ return Ok(Err(context.missing(&field.name)));
+ }
+ None => Datum::Null,
+ };
+ row.set_field(index, datum);
+ }
+ Ok(Ok(row))
+}
+
+struct ValueSeed<'a> {
+ path: ValuePath<'a>,
+ data_type: &'a DataType,
+}
+
+impl<'de> DeserializeSeed<'de> for ValueSeed<'_> {
+ type Value = DecodeResult<Datum<'static>>;
+
+ fn deserialize<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
+ where
+ D: de::Deserializer<'de>,
+ {
+ if is_complex(self.data_type) {
+ deserializer.deserialize_any(ComplexValueVisitor {
+ path: self.path,
+ data_type: self.data_type,
+ })
+ } else {
+ let raw = <&RawValue>::deserialize(deserializer)?;
+ let value = parse_scalar_value(raw).map_err(de::Error::custom)?;
+ Ok(decode_scalar(&self.path, self.data_type, &value))
+ }
+ }
+}
+
+struct ComplexValueVisitor<'a> {
+ path: ValuePath<'a>,
+ data_type: &'a DataType,
+}
+
+impl ComplexValueVisitor<'_> {
+ fn type_error(&self, actual: &str) -> DecodeResult<Datum<'static>> {
+ Err(self
+ .path
+ .type_error(complex_expected(self.data_type), actual))
+ }
+
+ fn null(&self) -> DecodeResult<Datum<'static>> {
+ if self.data_type.is_nullable() {
+ Ok(Datum::Null)
+ } else {
+ Err(self.path.invalid("must not be null"))
+ }
+ }
+}
+
+impl<'de> Visitor<'de> for ComplexValueVisitor<'_> {
+ type Value = DecodeResult<Datum<'static>>;
+
+ fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str(complex_expected(self.data_type))
+ }
+
+ fn visit_unit<E>(self) -> Result<Self::Value, E> {
+ Ok(self.null())
+ }
+
+ fn visit_bool<E>(self, _value: bool) -> Result<Self::Value, E> {
+ Ok(self.type_error("a boolean"))
+ }
+
+ fn visit_i64<E>(self, _value: i64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_u64<E>(self, _value: u64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_f64<E>(self, _value: f64) -> Result<Self::Value, E> {
+ Ok(self.type_error("a number"))
+ }
+
+ fn visit_str<E>(self, _value: &str) -> Result<Self::Value, E> {
+ Ok(self.type_error("a string"))
+ }
+
+ fn visit_seq<A>(self, sequence: A) -> Result<Self::Value, A::Error>
+ where
+ A: SeqAccess<'de>,
+ {
+ match self.data_type {
+ DataType::Array(data_type) => {
+ decode_array_sequence(sequence, &self.path, data_type.get_element_type())
+ }
+ DataType::Map(data_type) => decode_map_sequence(
+ sequence,
+ &self.path,
+ data_type.key_type(),
+ data_type.value_type(),
+ ),
+ DataType::Row(_) => {
+ let mut sequence = sequence;
+ drain_sequence(&mut sequence)?;
+ Ok(self.type_error("an array"))
+ }
+ _ => unreachable!("complex visitor receives only ARRAY, MAP, or ROW"),
+ }
+ }
+
+ fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>
+ where
+ A: MapAccess<'de>,
+ {
+ match self.data_type {
+ DataType::Row(row_type) => {
+ let row = decode_row_map(map, row_type, RowContext::Nested { path: self.path })?;
+ Ok(row.map(|row| Datum::Row(Box::new(row))))
+ }
+ DataType::Array(_) | DataType::Map(_) => {
+ let mut map = map;
+ drain_map(&mut map)?;
+ Ok(self.type_error("an object"))
+ }
+ _ => unreachable!("complex visitor receives only ARRAY, MAP, or ROW"),
+ }
+ }
+}
+
+fn complex_expected(data_type: &DataType) -> &'static str {
+ match data_type {
+ DataType::Array(_) => "ARRAY (a JSON array)",
+ DataType::Map(_) => "MAP (an array of {key, value} objects)",
Review Comment:
A MAP has to be [{"key":"a","value":1}], not {"a":1}. Makes sense for non-string keys and for keeping order, but most maps are string-keyed and the object form is a lot easier to write.
Should we accept both when K is a string?
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FIP-49][gateway] Add schema-aware JSON and Arrow row codecs for FIP-49 data types [fluss]
Posted by "naivedogger (via GitHub)" <gi...@apache.org>.
naivedogger commented on PR #4104:
URL: https://github.com/apache/fluss/pull/4104#issuecomment-5422784064
@beryllw @fresh-borzoni @leonardBang, appreciate a review here, thanks!
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org