JingsongLi commented on code in PR #69:
URL: https://github.com/apache/paimon-mosaic/pull/69#discussion_r3828951696
##########
cli/src/main.rs:
##########
@@ -518,36 +1343,1623 @@ fn convert(
Ok(())
}
-fn cat(
- file: &Path,
- num: usize,
- columns: Option<String>,
- filter: Option<String>,
- json: bool,
-) -> std::io::Result<()> {
- let mut reader = open(file)?;
- let pred = filter
- .as_deref()
- .map(filter::parse_where)
- .transpose()
- .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput,
e))?;
- let pred_col = match &pred {
- Some(p) => Some(
- reader
- .schema()
- .columns
- .iter()
- .position(|c| c.name == p.column)
- .ok_or_else(|| {
- std::io::Error::new(
- std::io::ErrorKind::InvalidInput,
- format!("--where: column '{}' not found", p.column),
- )
- })?,
- ),
- None => None,
- };
- // The display columns; the filter column is read even if projected out,
then
+fn project_convert_schema(schema: Schema, columns: &[String]) ->
std::io::Result<Schema> {
+ if columns.is_empty() {
+ return Ok(schema);
+ }
+ let mut seen = std::collections::HashSet::new();
+ let mut fields = Vec::new();
+ for name in columns {
+ if name.is_empty() {
+ return Err(invalid_schema("--column field name cannot be empty"));
+ }
+ let index = schema
+ .index_of(name)
+ .map_err(|_| invalid_schema(format!("--column '{name}' not found
in schema")))?;
+ if seen.insert(index) {
+ fields.push(schema.fields()[index].as_ref().clone());
+ }
+ }
+ Ok(Schema::new_with_metadata(fields, schema.metadata().clone()))
+}
+
+fn parse_convert_columns(arguments: &[String]) -> std::io::Result<Vec<String>>
{
+ if arguments.is_empty() {
+ return Ok(Vec::new());
+ }
+ let columns: Vec<String> = arguments
+ .iter()
+ .flat_map(|argument| parse_comma_list(argument))
+ .collect();
+ if columns.is_empty() {
+ return Err(invalid_schema("--column field name cannot be empty"));
+ }
+ Ok(columns)
+}
+
+fn infer_projected_json_schema<R: std::io::Read>(
+ reader: R,
+ columns: &[String],
+) -> Result<Schema, arrow::error::ArrowError> {
+ use arrow::error::ArrowError;
+
+ let values = serde_json::Deserializer::from_reader(reader)
+ .into_iter::<Value>()
+ .map(|value| {
+ let value = value.map_err(|e|
ArrowError::JsonError(e.to_string()))?;
+ Ok(match value {
+ Value::Object(mut object) => {
+ let projected = columns
+ .iter()
+ .filter_map(|name| object.remove(name).map(|value|
(name.clone(), value)))
+ .collect();
+ Value::Object(projected)
+ }
+ value => value,
+ })
+ });
+ arrow::json::reader::infer_json_schema_from_iterator(values)
+}
+
+/// Mosaic cannot store Arrow `Null` columns, and JSON inference produces
+/// `Null` for a field with no non-null value in the input — fail
+/// with the column name instead of the writer's late "unsupported DataType".
+fn reject_null_inferred_fields(schema: &Schema) -> std::io::Result<()> {
+ for field in schema.fields() {
+ if matches!(field.data_type(), DataType::Null) {
+ return Err(invalid_schema(format!(
+ "cannot infer a type for column '{}' (no non-null value in the
records); provide --schema",
+ fmt::safe(field.name())
+ )));
+ }
+ }
+ Ok(())
+}
+
+fn is_json_input(input: &Path) -> bool {
+ input
+ .extension()
+ .and_then(|ext| ext.to_str())
+ .is_some_and(|ext| {
+ matches!(
+ ext.to_ascii_lowercase().as_str(),
+ "json" | "ndjson" | "jsonl"
+ )
+ })
+}
+
+fn ensure_can_write(out: &Path, overwrite: bool) -> std::io::Result<()> {
+ if out.exists() && !overwrite {
+ return Err(std::io::Error::new(
+ std::io::ErrorKind::AlreadyExists,
+ format!("{} exists (use --overwrite to replace)", out.display()),
+ ));
+ }
+ Ok(())
+}
+
+fn csv_format(options: &CsvConvertOptions) ->
std::io::Result<arrow::csv::reader::Format> {
+ let delimiter = parse_csv_byte(&options.delimiter, "delimiter")?;
+ let escape = parse_optional_csv_byte(options.escape.as_deref(), "escape")?;
+ let quote = parse_csv_byte(&options.quote, "quote")?;
+ let format = arrow::csv::reader::Format::default()
+ .with_header(!options.no_header && options.header.is_none())
+ .with_delimiter(delimiter)
+ .with_quote(quote);
+ Ok(match escape {
+ Some(escape) => format.with_escape(escape),
+ None => format,
+ })
+}
+
+fn parse_csv_byte(value: &str, name: &str) -> std::io::Result<u8> {
+ let bytes = value.as_bytes();
+ if bytes.len() == 1 {
+ Ok(bytes[0])
+ } else {
+ Err(std::io::Error::new(
+ std::io::ErrorKind::InvalidInput,
+ format!("--{name} must be exactly one byte"),
+ ))
+ }
+}
+
+fn parse_optional_csv_byte(value: Option<&str>, name: &str) ->
std::io::Result<Option<u8>> {
+ value.map(|value| parse_csv_byte(value, name)).transpose()
+}
+
+fn open_csv(path: &Path, skip_lines: usize) ->
std::io::Result<std::io::BufReader<std::fs::File>> {
+ use std::io::BufRead;
+ let mut reader = std::io::BufReader::new(std::fs::File::open(path)?);
+ let mut line = String::new();
+ for _ in 0..skip_lines {
+ line.clear();
+ if reader.read_line(&mut line)? == 0 {
+ break;
+ }
+ }
+ Ok(reader)
+}
+
+struct CsvInputLayout {
+ header: Option<Vec<String>>,
+ columns: usize,
+ has_records: bool,
+}
+
+struct CsvInput {
+ reader: csv::Reader<std::io::BufReader<std::fs::File>>,
+ layout: CsvInputLayout,
+ first_record: Option<csv::StringRecord>,
+}
+
+const DEFAULT_CSV_BATCH_SIZE: usize = 1024;
+// Arrow's CSV RecordDecoder reserves roughly one data byte range and one
+// offset per cell before projection is applied. Keep that eager allocation
+// bounded for very wide records by reducing the number of rows per batch.
+const TARGET_CSV_DECODE_CELLS: usize = 64 * 1024;
+
+fn csv_batch_size(columns: usize) -> usize {
+ if columns == 0 {
+ return DEFAULT_CSV_BATCH_SIZE;
+ }
+ (TARGET_CSV_DECODE_CELLS / columns).clamp(1, DEFAULT_CSV_BATCH_SIZE)
+}
+
+fn explicit_csv_row_cells(source_columns: usize, output_columns: usize) ->
usize {
+ source_columns.max(output_columns)
+}
+
+fn csv_input_layout(path: &Path, options: &CsvConvertOptions) ->
std::io::Result<CsvInputLayout> {
+ Ok(open_csv_input(path, options)?.layout)
+}
+
+fn open_csv_input(path: &Path, options: &CsvConvertOptions) ->
std::io::Result<CsvInput> {
+ let delimiter = parse_csv_byte(&options.delimiter, "delimiter")?;
+ let escape = parse_optional_csv_byte(options.escape.as_deref(), "escape")?;
+ let quote = parse_csv_byte(&options.quote, "quote")?;
+ let mut builder = csv::ReaderBuilder::new();
+ builder
+ .has_headers(false)
+ .flexible(true)
+ .delimiter(delimiter)
+ .quote(quote)
+ .escape(escape);
+ let mut reader = builder.from_reader(open_csv(path, options.skip_lines)?);
+ let file_header = options.header.is_none() && !options.no_header;
+ let header = if let Some(header) = &options.header {
+ Some(parse_csv_header(header, options)?)
+ } else if options.no_header {
+ None
+ } else {
+ let mut record = csv::StringRecord::new();
+ if reader
+ .read_record(&mut record)
+ .map_err(|e| invalid_schema(format!("invalid CSV header: {e}")))?
+ {
+ Some(record.iter().map(ToString::to_string).collect())
+ } else {
+ Some(Vec::new())
+ }
+ };
+ let columns = header.as_ref().map_or(0, Vec::len);
+ let mut first_record = csv::StringRecord::new();
+ let has_records = reader
+ .read_record(&mut first_record)
+ .map_err(|e| invalid_schema(format!("invalid CSV record: {e}")))?;
+ if has_records && file_header {
+ validate_csv_header_names(header.as_ref().unwrap())?;
+ }
+ Ok(CsvInput {
+ reader,
+ layout: CsvInputLayout {
+ header,
+ columns,
+ has_records,
+ },
+ first_record: has_records.then_some(first_record),
+ })
+}
+
+fn write_explicit_schema_csv_input(
+ writer: &mut
paimon_mosaic_core::writer::MosaicWriter<paimon_mosaic_core::writer::FileSink>,
+ rows: &mut usize,
+ input: &Path,
+ schema: &Schema,
+ schema_index: &std::collections::HashMap<&str, usize>,
+ options: &CsvConvertOptions,
+) -> std::io::Result<()> {
+ let mut input_reader = open_csv_input(input, options)?;
+ if !input_reader.layout.has_records {
+ return Ok(());
+ }
+ let source_mapping = csv_output_mapping(schema, schema_index,
&input_reader.layout);
+ validate_csv_mapping(schema, &input_reader.layout, &source_mapping,
input)?;
+
+ let first = input_reader.first_record.take().into_iter().map(Ok);
+ let rest = std::iter::from_fn(|| {
+ let mut record = csv::StringRecord::new();
+ match input_reader.reader.read_record(&mut record) {
+ Ok(true) => Some(Ok(record)),
+ Ok(false) => None,
+ Err(e) => Some(Err(invalid_schema(format!(
+ "invalid CSV record in {}: {e}",
+ input.display()
+ )))),
+ }
+ });
+ for_each_explicit_csv_batch(
+ first.chain(rest),
+ input_reader.layout.columns,
+ schema,
+ TARGET_CONVERT_BATCH_BYTES,
+ |records| write_explicit_csv_records(writer, rows, schema,
&source_mapping, records),
+ )
+}
+
+fn for_each_explicit_csv_batch<I, F>(
+ records: I,
+ source_columns: usize,
+ schema: &Schema,
+ byte_budget: usize,
+ mut write: F,
+) -> std::io::Result<()>
+where
+ I: IntoIterator<Item = std::io::Result<csv::StringRecord>>,
+ F: FnMut(&[csv::StringRecord]) -> std::io::Result<()>,
+{
+ let output_columns = schema.fields().len();
+ let mut batch =
Vec::with_capacity(csv_batch_size(source_columns.max(output_columns)));
+ let mut cells: usize = 0;
+ let mut bytes: usize = 0;
+ for record in records {
+ let record = record?;
+ let row_cells = explicit_csv_row_cells(record.len(), output_columns);
+ let row_bytes = record.as_slice().len();
+ if !batch.is_empty()
+ && (batch.len() >= DEFAULT_CSV_BATCH_SIZE
+ || cells.saturating_add(row_cells) > TARGET_CSV_DECODE_CELLS
+ || bytes.saturating_add(row_bytes) > byte_budget)
+ {
+ write(&batch)?;
+ batch.clear();
+ cells = 0;
+ bytes = 0;
+ }
+ cells = cells.saturating_add(row_cells);
+ bytes = bytes.saturating_add(row_bytes);
+ batch.push(record);
+ if row_bytes > byte_budget {
+ write(&batch)?;
+ batch.clear();
+ cells = 0;
+ bytes = 0;
+ }
+ }
+ if !batch.is_empty() {
+ write(&batch)?;
+ }
+ Ok(())
+}
+
+fn write_explicit_csv_records(
+ writer: &mut
paimon_mosaic_core::writer::MosaicWriter<paimon_mosaic_core::writer::FileSink>,
+ rows: &mut usize,
+ schema: &Schema,
+ mapping: &[Option<usize>],
+ records: &[csv::StringRecord],
+) -> std::io::Result<()> {
+ let batch = csv_records_to_batch(schema, mapping, records)?;
+ *rows += batch.num_rows();
+ writer.write_batch(&batch)
+}
+
+fn csv_records_to_batch(
+ schema: &Schema,
+ mapping: &[Option<usize>],
+ records: &[csv::StringRecord],
+) -> std::io::Result<RecordBatch> {
+ let columns = schema
+ .fields()
+ .iter()
+ .zip(mapping)
+ .map(|(field, source)| match source {
+ Some(source) => csv_column_array(records, *source, field),
+ None => Ok(new_null_array(field.data_type(), records.len())),
+ })
+ .collect::<std::io::Result<Vec<_>>>()?;
+ RecordBatch::try_new(Arc::new(schema.clone()), columns)
+ .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData,
e.to_string()))
+}
+
+fn csv_column_array(
+ records: &[csv::StringRecord],
+ source: usize,
+ field: &Field,
+) -> std::io::Result<ArrayRef> {
+ match field.data_type() {
+ DataType::Boolean => {
+ let values = records
+ .iter()
+ .map(|record| {
+ let Some(value) = csv_record_value(record, source) else {
+ return Ok(None);
+ };
+ if value.eq_ignore_ascii_case("true") {
+ Ok(Some(true))
+ } else if value.eq_ignore_ascii_case("false") {
+ Ok(Some(false))
+ } else {
+ Err(csv_value_parse_error(record, field, value))
+ }
+ })
+ .collect::<std::io::Result<Vec<_>>>()?;
+ Ok(Arc::new(BooleanArray::from(values)))
+ }
+ DataType::Int32 => csv_primitive_column::<Int32Type>(records, source,
field),
+ DataType::Int64 => csv_primitive_column::<Int64Type>(records, source,
field),
+ DataType::Float32 => csv_primitive_column::<Float32Type>(records,
source, field),
+ DataType::Float64 => csv_primitive_column::<Float64Type>(records,
source, field),
+ DataType::Date32 => csv_primitive_column::<Date32Type>(records,
source, field),
+ DataType::Time32(TimeUnit::Millisecond) => {
+ csv_primitive_column::<Time32MillisecondType>(records, source,
field)
+ }
+ DataType::Timestamp(unit, timezone) => {
+ csv_timestamp_column(records, source, field, unit,
timezone.clone())
+ }
+ DataType::Decimal128(precision, scale) => {
+ let values = records
+ .iter()
+ .map(|record| {
+ csv_record_value(record, source)
+ .map(|value| match parse_decimal_unscaled_exact(
+ value, *precision, *scale,
+ ) {
+ Ok(parsed) => Ok(Some(parsed)),
+ Err(DecimalParseFailure::Inexact) => {
+ Err(invalid_schema(format!(
+ "decimal value '{}' for CSV field '{}' at
line {} cannot be represented exactly with scale {scale}",
+ fmt::safe(value),
+ fmt::safe(field.name()),
+ record
+ .position()
+ .map(|position|
position.line().to_string())
+ .unwrap_or_else(||
"unknown".to_string())
+ )))
+ }
+ Err(_) => Err(csv_value_parse_error(record, field,
value)),
+ })
+ .unwrap_or(Ok(None))
+ })
+ .collect::<std::io::Result<Vec<_>>>()?;
+ let array: PrimitiveArray<Decimal128Type> =
values.into_iter().collect();
+ Ok(Arc::new(
+ array
+ .with_precision_and_scale(*precision, *scale)
+ .map_err(|e| invalid_schema(e.to_string()))?,
+ ))
+ }
+ DataType::Utf8 => {
+ let values = records
+ .iter()
+ .map(|record| {
+ let value = csv_record_value(record, source);
+ if let Some(value) = value {
+ if field_is_avro_uuid(field) &&
validate_avro_uuid(value).is_err() {
+ return Err(invalid_schema(format!(
+ "invalid UUID '{}' for CSV field '{}' at line
{}",
+ fmt::safe(value),
+ fmt::safe(field.name()),
+ record
+ .position()
+ .map(|position|
position.line().to_string())
+ .unwrap_or_else(|| "unknown".to_string())
+ )));
+ }
+ }
+ Ok(value)
+ })
+ .collect::<std::io::Result<StringArray>>()?;
+ Ok(Arc::new(values))
+ }
+ data_type => Err(invalid_schema(format!(
+ "CSV conversion does not support field '{}' with type {data_type}",
+ fmt::safe(field.name())
+ ))),
+ }
+}
+
+fn csv_primitive_column<T>(
+ records: &[csv::StringRecord],
+ source: usize,
+ field: &Field,
+) -> std::io::Result<ArrayRef>
+where
+ T: ArrowPrimitiveType + ArrowValueParser,
+{
+ Ok(Arc::new(csv_primitive_array::<T>(records, source, field)?))
+}
+
+fn csv_primitive_array<T>(
+ records: &[csv::StringRecord],
+ source: usize,
+ field: &Field,
+) -> std::io::Result<PrimitiveArray<T>>
+where
+ T: ArrowPrimitiveType + ArrowValueParser,
+{
+ let values = records
+ .iter()
+ .map(|record| {
+ csv_record_value(record, source)
+ .map(|value| {
+ T::parse(value)
+ .map(Some)
+ .ok_or_else(|| csv_value_parse_error(record, field,
value))
+ })
+ .unwrap_or(Ok(None))
+ })
+ .collect::<std::io::Result<Vec<_>>>()?;
+ Ok(values.into_iter().collect())
+}
+
+fn csv_timestamp_column(
+ records: &[csv::StringRecord],
+ source: usize,
+ field: &Field,
+ unit: &TimeUnit,
+ timezone: Option<Arc<str>>,
+) -> std::io::Result<ArrayRef> {
+ let parser_timezone: Tz = timezone
+ .as_deref()
+ .unwrap_or("+00:00")
+ .parse()
+ .map_err(|e| invalid_schema(format!("invalid timestamp timezone:
{e}")))?;
+ let timezone_policy = timezone.is_none().then_some("a local timestamp");
+ let values = records
+ .iter()
+ .map(|record| {
+ let Some(value) = csv_record_value(record, source) else {
+ return Ok(None);
+ };
+ let line = record
+ .position()
+ .map(|position| position.line().to_string())
+ .unwrap_or_else(|| "unknown".to_string());
+ parse_csv_timestamp_value(
+ value,
+ field,
+ unit,
+ &parser_timezone,
+ timezone_policy,
+ &format!("at line {line}"),
+ )
+ .map(Some)
+ })
+ .collect::<std::io::Result<Vec<_>>>()?;
+ csv_timestamp_array(values, unit, timezone)
+}
+
+fn parse_csv_timestamp_value(
+ value: &str,
+ field: &Field,
+ unit: &TimeUnit,
+ parser_timezone: &Tz,
+ timezone_policy: Option<&str>,
+ location: &str,
+) -> std::io::Result<i64> {
+ match timezone_policy {
+ Some(policy) if timestamp_has_explicit_timezone(value) => {
+ return Err(invalid_schema(format!(
+ "CSV field '{}' {location} must not include a timezone for
{policy}",
+ fmt::safe(field.name())
+ )));
+ }
+ _ => {}
+ }
+ let parse_error = || {
+ invalid_schema(format!(
+ "cannot parse '{}' as {} for CSV field '{}' {location}",
+ fmt::safe(value),
+ field.data_type(),
+ fmt::safe(field.name())
+ ))
+ };
+ let datetime = string_to_datetime(parser_timezone, value).map_err(|_|
parse_error())?;
+ match unit {
+ TimeUnit::Millisecond => Ok(datetime.timestamp_millis()),
+ TimeUnit::Microsecond => Ok(datetime.timestamp_micros()),
+ TimeUnit::Nanosecond =>
datetime.timestamp_nanos_opt().ok_or_else(parse_error),
+ unit => Err(invalid_schema(format!(
+ "CSV conversion does not support timestamp unit {unit:?}"
+ ))),
+ }
+}
+
+fn csv_timestamp_array(
+ values: Vec<Option<i64>>,
+ unit: &TimeUnit,
+ timezone: Option<Arc<str>>,
+) -> std::io::Result<ArrayRef> {
+ Ok(match unit {
+ TimeUnit::Millisecond => Arc::new(
+ PrimitiveArray::<TimestampMillisecondType>::from(values)
+ .with_timezone_opt(timezone.clone()),
+ ),
+ TimeUnit::Microsecond => Arc::new(
+ PrimitiveArray::<TimestampMicrosecondType>::from(values)
+ .with_timezone_opt(timezone.clone()),
+ ),
+ TimeUnit::Nanosecond => Arc::new(
+
PrimitiveArray::<TimestampNanosecondType>::from(values).with_timezone_opt(timezone),
+ ),
+ unit => {
+ return Err(invalid_schema(format!(
+ "CSV conversion does not support timestamp unit {unit:?}"
+ )));
+ }
+ })
+}
+
+fn parse_decimal_exact(
+ value: &str,
+ precision: u8,
+ scale: i8,
+) -> Result<i128, arrow::error::ArrowError> {
+ parse_decimal_unscaled_exact(value, precision, scale).map_err(|failure| {
+ let message = match failure {
+ DecimalParseFailure::Invalid => {
+ format!("can't parse the string value {value} to decimal")
+ }
+ DecimalParseFailure::Inexact => {
+ format!("cannot be represented exactly with scale {scale}")
+ }
+ DecimalParseFailure::Overflow => format!("parse decimal overflow
({value})"),
+ };
+ arrow::error::ArrowError::ParseError(message)
+ })
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum DecimalParseFailure {
+ Invalid,
+ Inexact,
+ Overflow,
+}
+
+fn parse_decimal_unscaled_exact(
+ value: &str,
+ precision: u8,
+ scale: i8,
+) -> Result<i128, DecimalParseFailure> {
+ if precision == 0 || precision > 38 {
+ return Err(DecimalParseFailure::Overflow);
+ }
+ let (negative, unsigned) = match value.as_bytes().first() {
+ Some(b'-') => (true, &value[1..]),
+ Some(b'+') => (false, &value[1..]),
+ _ => (false, value),
+ };
+ let exponent_index = unsigned.find(['e', 'E']);
+ let (mantissa, exponent) = match exponent_index {
+ Some(index) => {
+ let exponent = unsigned[index + 1..]
+ .parse::<i64>()
+ .map_err(|_| DecimalParseFailure::Invalid)?;
+ (&unsigned[..index], exponent)
+ }
+ None => (unsigned, 0),
+ };
+
+ let mut seen_decimal_point = false;
+ let mut has_digit = false;
+ let mut total_digits = 0_usize;
+ let mut fractional_digits = 0_usize;
+ let mut first_nonzero = None;
+ for byte in mantissa.bytes() {
+ match byte {
+ b'0'..=b'9' => {
+ has_digit = true;
+ if seen_decimal_point {
+ fractional_digits = fractional_digits
+ .checked_add(1)
+ .ok_or(DecimalParseFailure::Overflow)?;
+ }
+ if byte != b'0' && first_nonzero.is_none() {
+ first_nonzero = Some(total_digits);
+ }
+ total_digits = total_digits
+ .checked_add(1)
+ .ok_or(DecimalParseFailure::Overflow)?;
+ }
+ b'.' if !seen_decimal_point => seen_decimal_point = true,
+ _ => return Err(DecimalParseFailure::Invalid),
+ }
+ }
+ if !has_digit {
+ return Err(DecimalParseFailure::Invalid);
+ }
+ let Some(first_nonzero) = first_nonzero else {
+ return Ok(0);
+ };
+
+ let shift = i128::from(exponent) - fractional_digits as i128 +
i128::from(scale);
+ let (kept_digits, appended_zeros) = if shift >= 0 {
+ let appended_zeros = usize::try_from(shift).map_err(|_|
DecimalParseFailure::Overflow)?;
+ (total_digits, appended_zeros)
+ } else {
+ let discarded =
+ usize::try_from(shift.unsigned_abs()).map_err(|_|
DecimalParseFailure::Inexact)?;
+ if discarded > total_digits {
+ return Err(DecimalParseFailure::Inexact);
+ }
+ let kept_digits = total_digits - discarded;
+ for (digit_index, byte) in
mantissa.bytes().filter(u8::is_ascii_digit).enumerate() {
+ if digit_index >= kept_digits && byte != b'0' {
+ return Err(DecimalParseFailure::Inexact);
+ }
+ }
+ (kept_digits, 0)
+ };
+
+ let significant_digits = kept_digits
+ .saturating_sub(first_nonzero)
+ .checked_add(appended_zeros)
+ .ok_or(DecimalParseFailure::Overflow)?;
+ if significant_digits > usize::from(precision) {
+ return Err(DecimalParseFailure::Overflow);
+ }
+
+ let mut result = 0_i128;
+ for (digit_index, byte) in
mantissa.bytes().filter(u8::is_ascii_digit).enumerate() {
+ if (first_nonzero..kept_digits).contains(&digit_index) {
+ result = result
+ .checked_mul(10)
+ .and_then(|value| value.checked_add(i128::from(byte - b'0')))
+ .ok_or(DecimalParseFailure::Overflow)?;
+ }
+ }
+ for _ in 0..appended_zeros {
+ result = result
+ .checked_mul(10)
+ .ok_or(DecimalParseFailure::Overflow)?;
+ }
+ if negative {
+ result.checked_neg().ok_or(DecimalParseFailure::Overflow)
+ } else {
+ Ok(result)
+ }
+}
+
+fn timestamp_has_explicit_timezone(value: &str) -> bool {
+ // Arrow's timestamp parser requires a fixed-width YYYY-MM-DD date prefix
+ // and accepts colon-separated or compact time fields. The offsets below
+ // intentionally mirror that grammar; expanded-year forms fail parsing.
+ let bytes = value.trim().as_bytes();
+ if bytes.len() <= 10 {
+ return false;
+ }
+ let mut timezone_start = if bytes.get(13) == Some(&b':') && bytes.get(16)
== Some(&b':') {
+ 19
+ } else {
+ 17
+ };
+ if bytes.get(timezone_start) == Some(&b'.') {
+ timezone_start += 1;
+ while bytes.get(timezone_start).is_some_and(u8::is_ascii_digit) {
+ timezone_start += 1;
+ }
+ }
+ timezone_start < bytes.len()
+}
+
+fn csv_record_value(record: &csv::StringRecord, source: usize) -> Option<&str>
{
+ record.get(source).filter(|value| !value.is_empty())
+}
+
+fn csv_value_parse_error(record: &csv::StringRecord, field: &Field, value:
&str) -> std::io::Error {
+ let line = record
+ .position()
+ .map(|position| position.line().to_string())
+ .unwrap_or_else(|| "unknown".to_string());
+ std::io::Error::new(
+ std::io::ErrorKind::InvalidData,
+ format!(
+ "cannot parse '{}' as {} for CSV field '{}' at line {line}",
+ fmt::safe(value),
+ field.data_type(),
+ fmt::safe(field.name())
+ ),
+ )
+}
+
+fn csv_schema_index(schema: &Schema) -> std::collections::HashMap<&str, usize>
{
+ schema
+ .fields()
+ .iter()
+ .enumerate()
+ .map(|(index, field)| (field.name().as_str(), index))
+ .collect()
+}
+
+fn mixed_csv_float_fields(
+ output_schema: &Schema,
+ inferred_types: &std::collections::HashMap<String, ObservedCsvTypes>,
+) -> std::collections::HashSet<String> {
+ // These fields are Float64 only because different shards inferred Int64
+ // and Float64. Read their raw text in every shard so integer-looking
values
+ // cannot be rounded before the exactness check.
+ output_schema
+ .fields()
+ .iter()
+ .filter(|field| matches!(field.data_type(), DataType::Float64))
+ .filter(|field| {
+ inferred_types
+ .get(field.name())
+ .is_some_and(ObservedCsvTypes::is_mixed_int_float)
+ })
+ .map(|field| field.name().clone())
+ .collect()
+}
+
+#[derive(Default)]
+struct ObservedCsvTypes {
+ saw_int64: bool,
+ saw_float64: bool,
+}
+
+impl ObservedCsvTypes {
+ fn observe(&mut self, data_type: &DataType) {
+ self.saw_int64 |= matches!(data_type, DataType::Int64);
+ self.saw_float64 |= matches!(data_type, DataType::Float64);
+ }
+
+ fn is_mixed_int_float(&self) -> bool {
+ self.saw_int64 && self.saw_float64
+ }
+}
+
+fn observe_csv_inferred_types(
+ inferred_types: &mut std::collections::HashMap<String, ObservedCsvTypes>,
+ schema: &Schema,
+) {
+ for field in schema.fields() {
+ if matches!(field.data_type(), DataType::Int64 | DataType::Float64) {
Review Comment:
[P1] Preserve integer tokens inside Float64-inferred shards
This records only the final inferred type for each shard. If a shard
contains both 9007199254740993 and 1.5, Arrow already reports Float64, so
saw_int64 remains false and mixed_csv_float_fields does not route the column
through the raw-text exactness check. With another Float64 shard, conversion
succeeds but stores 9007199254740992, silently changing a valid Int64 token.
Please retain per-token integer observations during inference, or read every
inferred Float64 column as Utf8 and apply the existing exact-integral-token
check. A regression test should cover multiple inputs where no shard infers as
pure Int64.
##########
cli/src/main.rs:
##########
@@ -429,48 +530,777 @@ fn count(file: &Path, json: bool) -> std::io::Result<()>
{
fn convert(
input: &Path,
out: &Path,
- stats: Option<String>,
+ schema: Option<&Path>,
+ columns: &[String],
+ stats: Option<&str>,
overwrite: bool,
) -> std::io::Result<()> {
- if out.exists() && !overwrite {
+ use arrow::error::ArrowError;
+ let bad = |e: ArrowError|
std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string());
+ if !is_json_input(input) {
return Err(std::io::Error::new(
- std::io::ErrorKind::AlreadyExists,
- format!("{} exists (use --overwrite to replace)", out.display()),
+ std::io::ErrorKind::InvalidInput,
+ "convert only supports JSON inputs (.json/.ndjson/.jsonl); use
convert-csv for CSV data",
));
}
- use arrow::error::ArrowError;
- use paimon_mosaic_core::writer::{MosaicWriter, WriterOptions};
- let bad = |e: ArrowError|
std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string());
- let is_json = matches!(
- input.extension().and_then(|e| e.to_str()),
- Some("json") | Some("ndjson") | Some("jsonl")
- );
- // Infer schema, then build a batch iterator — CSV (header) or JSON (one
object per line).
- type Batches = Box<dyn Iterator<Item = Result<RecordBatch, ArrowError>>>;
- // Schema inference and the data reader each need their own pass over the
- // file (inference consumes a reader), so open it twice via one helper.
+ let columns = parse_convert_columns(columns)?;
+ ensure_can_write(out, overwrite)?;
+ let explicit_schema = schema.map(load_convert_schema).transpose()?;
let open =
|| -> std::io::Result<_> {
Ok(std::io::BufReader::new(std::fs::File::open(input)?)) };
- let (schema, reader): (arrow::datatypes::Schema, Batches) = if is_json {
- let (schema, _) =
- arrow::json::reader::infer_json_schema(&mut open()?,
None).map_err(bad)?;
- let rd =
arrow::json::ReaderBuilder::new(std::sync::Arc::new(schema.clone()))
- .build(open()?)
- .map_err(bad)?;
- (schema, Box::new(rd))
- } else {
- let (schema, _) = arrow::csv::reader::Format::default()
- .with_header(true)
- .infer_schema(open()?, None)
- .map_err(bad)?;
- let rd =
arrow::csv::ReaderBuilder::new(std::sync::Arc::new(schema.clone()))
- .with_header(true)
- .build(open()?)
- .map_err(bad)?;
- (schema, Box::new(rd))
+ let has_explicit_schema = explicit_schema.is_some();
+ let schema = match explicit_schema {
+ Some(schema) => schema,
+ None if columns.is_empty() =>
arrow::json::reader::infer_json_schema(&mut open()?, None)
+ .map(|(schema, _)| schema)
+ .map_err(bad)?,
+ None => infer_projected_json_schema(open()?, &columns).map_err(bad)?,
+ };
+ let schema = project_convert_schema(schema, &columns)?;
+ reject_null_inferred_fields(&schema)?;
+ reject_json_unsupported_fields(&schema)?;
+ if has_explicit_schema && schema_needs_json_validation(&schema) {
+ return write_mosaic(out, overwrite, &schema, stats, |writer, rows| {
+ write_validated_json_input(open()?, &schema, writer, rows)
+ });
+ }
+ let reader = arrow::json::ReaderBuilder::new(Arc::new(schema.clone()))
+ .build(open()?)
+ .map_err(bad)?;
+ write_mosaic(out, overwrite, &schema, stats, |writer, rows| {
+ for batch in reader {
+ let batch = batch
+ .map_err(|e|
std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?;
+ *rows += batch.num_rows();
+ writer.write_batch(&batch)?;
+ }
+ Ok(())
+ })
+}
+
+fn write_validated_json_input<R: std::io::BufRead>(
+ reader: R,
+ schema: &Schema,
+ writer: &mut
paimon_mosaic_core::writer::MosaicWriter<paimon_mosaic_core::writer::FileSink>,
+ rows: &mut usize,
+) -> std::io::Result<()> {
+ for_each_validated_json_batch(reader, schema, TARGET_CONVERT_BATCH_BYTES,
|batch| {
+ *rows += batch.num_rows();
+ writer.write_batch(&batch)
+ })
+}
+
+const DEFAULT_JSON_BATCH_SIZE: usize = 1024;
+const TARGET_CONVERT_BATCH_BYTES: usize = 16 * 1024 * 1024;
+
+fn for_each_validated_json_batch<R, F>(
+ reader: R,
+ schema: &Schema,
+ byte_budget: usize,
+ mut write: F,
+) -> std::io::Result<()>
+where
+ R: std::io::Read,
+ F: FnMut(RecordBatch) -> std::io::Result<()>,
+{
+ let bad = |e: arrow::error::ArrowError| {
+ std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string())
+ };
+ let build_decoder = || {
+ arrow::json::ReaderBuilder::new(Arc::new(schema.clone()))
+ .with_batch_size(DEFAULT_JSON_BATCH_SIZE)
+ .build_decoder()
+ .map_err(bad)
+ };
+ let fields = json_special_fields(schema);
+ let normalize_decimals = schema_has_decimal(schema);
+ let mut decoder = build_decoder()?;
+ let mut batch_bytes = 0_usize;
+ let records =
serde_json::Deserializer::from_reader(reader).into_iter::<Box<RawValue>>();
+
+ for (index, raw) in records.enumerate() {
+ let record = index + 1;
+ let raw = raw.map_err(|e| invalid_schema(format!("invalid JSON record
{record}: {e}")))?;
+ let raw_bytes = raw.get().as_bytes();
+ validate_json_special_values(raw_bytes, &fields, record)?;
+ let normalized;
+ let decode_bytes = if normalize_decimals {
+ normalized = normalize_json_decimal_record(&raw, schema, record)?;
+ normalized.as_bytes()
+ } else {
+ raw_bytes
+ };
+ if !decoder.is_empty()
+ && (decoder.len() >= DEFAULT_JSON_BATCH_SIZE
+ || batch_bytes.saturating_add(decode_bytes.len()) >
byte_budget)
+ {
+ if let Some(batch) = decoder.flush().map_err(bad)? {
+ write(batch)?;
+ }
+ batch_bytes = 0;
+ }
+
+ let decoded = decoder.decode(decode_bytes).map_err(bad)?;
+ if decoded != decode_bytes.len() || decoder.has_partial_record() {
+ return Err(invalid_schema(format!(
+ "invalid JSON record {record}: decoder stopped before the
record ended"
+ )));
+ }
+ batch_bytes = batch_bytes.saturating_add(decode_bytes.len());
+
+ if decoder.len() >= DEFAULT_JSON_BATCH_SIZE || batch_bytes >=
byte_budget {
+ if let Some(batch) = decoder.flush().map_err(bad)? {
+ write(batch)?;
+ }
+ if batch_bytes > byte_budget {
+ decoder = build_decoder()?;
+ }
+ batch_bytes = 0;
+ }
+ }
+
+ if let Some(batch) = decoder.flush().map_err(bad)? {
+ write(batch)?;
+ }
+ Ok(())
+}
+
+fn schema_has_decimal(schema: &Schema) -> bool {
+ schema
+ .fields()
+ .iter()
+ .any(|field| data_type_has_decimal(field.data_type()))
+}
+
+fn data_type_has_decimal(data_type: &DataType) -> bool {
+ match data_type {
+ DataType::Decimal128(_, _) => true,
+ DataType::List(field) => data_type_has_decimal(field.data_type()),
+ DataType::Map(entries, _) => match entries.data_type() {
+ DataType::Struct(fields) => fields
+ .get(1)
+ .is_some_and(|field| data_type_has_decimal(field.data_type())),
+ _ => false,
+ },
+ DataType::Struct(fields) => fields
+ .iter()
+ .any(|field| data_type_has_decimal(field.data_type())),
+ _ => false,
+ }
+}
+
+fn normalize_json_decimal_record(
+ raw: &RawValue,
+ schema: &Schema,
+ record: usize,
+) -> std::io::Result<String> {
+ let values: std::collections::BTreeMap<String, Box<RawValue>> =
serde_json::from_str(raw.get())
+ .map_err(|e| invalid_schema(format!("invalid JSON record {record}:
{e}")))?;
+ let mut normalized = String::from("{");
+ for (index, (name, value)) in values.iter().enumerate() {
+ if index != 0 {
+ normalized.push(',');
+ }
+ normalized.push_str(
+ &serde_json::to_string(name)
+ .map_err(|e| invalid_schema(format!("invalid JSON field name:
{e}")))?,
+ );
+ normalized.push(':');
+ if let Some(field) = schema.fields().iter().find(|field| field.name()
== name) {
+ normalized.push_str(&normalize_json_decimal_value(value, field,
name, record)?);
+ } else {
+ normalized.push_str(value.get());
+ }
+ }
+ normalized.push('}');
+ Ok(normalized)
+}
+
+fn normalize_json_decimal_value(
+ raw: &RawValue,
+ field: &Field,
+ path: &str,
+ record: usize,
+) -> std::io::Result<String> {
+ if raw.get() == "null" || !data_type_has_decimal(field.data_type()) {
+ return Ok(raw.get().to_string());
+ }
+ match field.data_type() {
+ DataType::Decimal128(precision, scale) => {
+ let raw_text = raw.get();
+ let value = if raw_text.starts_with('"') {
+ serde_json::from_str::<String>(raw_text)
+ .map_err(|e| invalid_schema(format!("invalid JSON decimal:
{e}")))?
+ } else {
+ raw_text.to_string()
+ };
+ let unscaled = parse_decimal_exact(&value, *precision,
*scale).map_err(|e| {
+ invalid_schema(format!(
+ "cannot parse '{}' as {} for JSON field '{}' at record
{record}: {e}",
+ fmt::safe(&value),
+ field.data_type(),
+ fmt::safe(path)
+ ))
+ })?;
+ serde_json::to_string(&format_decimal_unscaled(unscaled, *scale))
+ .map_err(|e| invalid_schema(format!("invalid JSON decimal:
{e}")))
+ }
+ DataType::List(item) => {
+ let values: Vec<Box<RawValue>> = serde_json::from_str(raw.get())
+ .map_err(|e| invalid_schema(format!("invalid JSON array:
{e}")))?;
+ let child_path = format!("{path}[]");
+ let mut normalized = String::from("[");
+ for (index, value) in values.iter().enumerate() {
+ if index != 0 {
+ normalized.push(',');
+ }
+ normalized.push_str(&normalize_json_decimal_value(
+ value,
+ item,
+ &child_path,
+ record,
+ )?);
+ }
+ normalized.push(']');
+ Ok(normalized)
+ }
+ DataType::Map(entries, _) => {
+ let DataType::Struct(fields) = entries.data_type() else {
+ return Ok(raw.get().to_string());
+ };
+ let Some(value_field) = fields.get(1) else {
+ return Ok(raw.get().to_string());
+ };
+ let values: std::collections::BTreeMap<String, Box<RawValue>> =
+ serde_json::from_str(raw.get())
+ .map_err(|e| invalid_schema(format!("invalid JSON map:
{e}")))?;
+ let child_path = format!("{path}{{}}");
+ let mut normalized = String::from("{");
+ for (index, (name, value)) in values.iter().enumerate() {
+ if index != 0 {
+ normalized.push(',');
+ }
+ normalized.push_str(
+ &serde_json::to_string(name)
+ .map_err(|e| invalid_schema(format!("invalid JSON map
key: {e}")))?,
+ );
+ normalized.push(':');
+ normalized.push_str(&normalize_json_decimal_value(
+ value,
+ value_field,
+ &child_path,
+ record,
+ )?);
+ }
+ normalized.push('}');
+ Ok(normalized)
+ }
+ DataType::Struct(fields) => {
+ let values: std::collections::BTreeMap<String, Box<RawValue>> =
+ serde_json::from_str(raw.get())
+ .map_err(|e| invalid_schema(format!("invalid JSON object:
{e}")))?;
+ let mut normalized = String::from("{");
+ for (index, (name, value)) in values.iter().enumerate() {
+ if index != 0 {
+ normalized.push(',');
+ }
+ normalized.push_str(
+ &serde_json::to_string(name)
+ .map_err(|e| invalid_schema(format!("invalid JSON
field name: {e}")))?,
+ );
+ normalized.push(':');
+ if let Some(child) = fields.iter().find(|field| field.name()
== name) {
+ normalized.push_str(&normalize_json_decimal_value(
+ value,
+ child,
+ &format!("{path}.{name}"),
+ record,
+ )?);
+ } else {
+ normalized.push_str(value.get());
+ }
+ }
+ normalized.push('}');
+ Ok(normalized)
+ }
+ _ => Ok(raw.get().to_string()),
+ }
+}
+
+fn format_decimal_unscaled(unscaled: i128, scale: i8) -> String {
+ let negative = unscaled < 0;
+ let mut digits = unscaled.unsigned_abs().to_string();
+ if scale > 0 {
+ let scale = scale as usize;
+ if digits.len() <= scale {
+ digits.insert_str(0, &"0".repeat(scale + 1 - digits.len()));
+ }
+ digits.insert(digits.len() - scale, '.');
+ } else if scale < 0 {
+ digits.push_str(&"0".repeat(scale.unsigned_abs() as usize));
+ }
+ if negative {
+ digits.insert(0, '-');
+ }
+ digits
+}
+
+fn schema_needs_json_validation(schema: &Schema) -> bool {
+ schema
+ .fields()
+ .iter()
+ .any(|field| field_needs_json_validation(field))
+}
+
+const AVRO_LOGICAL_TYPE_METADATA: &str = "paimon.mosaic.avro.logical_type";
+const AVRO_UUID_LOGICAL_TYPE: &str = "uuid";
+
+fn field_needs_json_validation(field: &Field) -> bool {
+ field_is_avro_uuid(field) ||
data_type_needs_json_validation(field.data_type())
+}
+
+fn field_is_avro_uuid(field: &Field) -> bool {
+ field
+ .metadata()
+ .get(AVRO_LOGICAL_TYPE_METADATA)
+ .is_some_and(|value| value == AVRO_UUID_LOGICAL_TYPE)
+}
+
+fn data_type_needs_json_validation(data_type: &DataType) -> bool {
+ match data_type {
+ DataType::Int32
+ | DataType::Int64
+ | DataType::Date32
+ | DataType::Time32(_)
+ | DataType::Timestamp(_, _)
+ | DataType::Decimal128(_, _) => true,
+ DataType::List(field) => field_needs_json_validation(field),
+ // Every map must be walked so duplicate keys are rejected even when
+ // its values do not otherwise need special Avro validation.
+ DataType::Map(_, _) => true,
+ DataType::Struct(fields) => fields
+ .iter()
+ .any(|field| field_needs_json_validation(field)),
+ _ => false,
+ }
+}
+
+fn validate_json_special_values(
+ raw: &[u8],
+ fields: &std::collections::HashMap<String, Arc<Field>>,
+ first_record: usize,
+) -> std::io::Result<()> {
+ // Borrow only the raw values of relevant fields. Unrelated values are
+ // skipped without constructing a second set of Arrow arrays or a Value
tree.
+ let mut deserializer = serde_json::Deserializer::from_slice(raw);
+ JsonSpecialRecordSeed {
+ fields,
+ record: first_record,
+ }
+ .deserialize(&mut deserializer)
+ .map_err(|e| invalid_schema(format!("invalid JSON record {first_record}:
{e}")))
+}
+
+fn json_special_fields(schema: &Schema) -> std::collections::HashMap<String,
Arc<Field>> {
+ schema
+ .fields()
+ .iter()
+ .filter(|field| field_needs_json_validation(field))
+ .map(|field| (field.name().clone(), Arc::clone(field)))
+ .collect()
+}
+
+struct JsonSpecialRecordSeed<'a> {
+ fields: &'a std::collections::HashMap<String, Arc<Field>>,
+ record: usize,
+}
+
+impl<'de> DeserializeSeed<'de> for JsonSpecialRecordSeed<'_> {
+ type Value = ();
+
+ fn deserialize<D>(self, deserializer: D) -> Result<(), D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ deserializer.deserialize_map(JsonSpecialRecordVisitor {
+ fields: self.fields,
+ record: self.record,
+ })
+ }
+}
+
+struct JsonSpecialRecordVisitor<'a> {
+ fields: &'a std::collections::HashMap<String, Arc<Field>>,
+ record: usize,
+}
+
+impl<'de> Visitor<'de> for JsonSpecialRecordVisitor<'_> {
+ type Value = ();
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) ->
std::fmt::Result {
+ formatter.write_str("a JSON object")
+ }
+
+ fn visit_map<M>(self, mut map: M) -> Result<(), M::Error>
+ where
+ M: MapAccess<'de>,
+ {
+ while let Some(name) = map.next_key::<std::borrow::Cow<'de, str>>()? {
+ if let Some(field) = self.fields.get(name.as_ref()) {
+ let raw: &RawValue = map.next_value()?;
+ validate_json_special_value(raw, field, name.as_ref(),
self.record)
+ .map_err(M::Error::custom)?;
+ } else {
+ map.next_value::<IgnoredAny>()?;
+ }
+ }
+ Ok(())
+ }
+}
+
+struct JsonSpecialMapSeed<'a> {
+ value_field: &'a Field,
+ path: &'a str,
+ record: usize,
+}
+
+impl<'de> DeserializeSeed<'de> for JsonSpecialMapSeed<'_> {
+ type Value = ();
+
+ fn deserialize<D>(self, deserializer: D) -> Result<(), D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ deserializer.deserialize_map(JsonSpecialMapVisitor {
+ value_field: self.value_field,
+ path: self.path,
+ record: self.record,
+ })
+ }
+}
+
+struct JsonSpecialMapVisitor<'a> {
+ value_field: &'a Field,
+ path: &'a str,
+ record: usize,
+}
+
+impl<'de> Visitor<'de> for JsonSpecialMapVisitor<'_> {
+ type Value = ();
+
+ fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) ->
std::fmt::Result {
+ formatter.write_str("a JSON map")
+ }
+
+ fn visit_map<M>(self, mut map: M) -> Result<(), M::Error>
+ where
+ M: MapAccess<'de>,
+ {
+ let mut seen = std::collections::HashSet::new();
+ while let Some(key) = map.next_key::<std::borrow::Cow<'de, str>>()? {
+ if !seen.insert(key.to_string()) {
+ return Err(M::Error::custom(format!(
+ "duplicate JSON map key '{}' in field '{}' at record {}",
+ fmt::safe(key.as_ref()),
+ fmt::safe(self.path),
+ self.record
+ )));
+ }
+ let raw: &RawValue = map.next_value()?;
+ validate_json_special_value(raw, self.value_field, self.path,
self.record)
+ .map_err(M::Error::custom)?;
+ }
+ Ok(())
+ }
+}
+
+fn validate_json_special_value(
+ raw: &RawValue,
+ field: &Field,
+ path: &str,
+ record: usize,
+) -> std::io::Result<()> {
+ if raw.get() == "null" || !field_needs_json_validation(field) {
+ return Ok(());
+ }
+ if field_is_avro_uuid(field) {
+ let value: String = serde_json::from_str(raw.get()).map_err(|_| {
+ invalid_schema(format!(
+ "JSON field '{}' at record {record} must be a valid UUID
string",
+ fmt::safe(path)
+ ))
+ })?;
+ validate_avro_uuid(&value).map_err(|_| {
+ invalid_schema(format!(
+ "invalid UUID '{}' for JSON field '{}' at record {record}",
+ fmt::safe(&value),
+ fmt::safe(path)
+ ))
+ })?;
+ }
+ let data_type = field.data_type();
+ match data_type {
+ DataType::Int32 | DataType::Date32 | DataType::Time32(_) if
!raw.get().starts_with('"') => {
Review Comment:
[P1] Enforce the Avro time-millis day range
Time32(Millisecond) is validated only against i32 bounds here, but Avro
time-millis is milliseconds after midnight, so valid values are 0 <= value <
86400000. Inputs -1 and 86400000 currently convert successfully and persist
invalid Time32 values; cat --json later emits a temporal cast error. Please
validate this type separately and add boundary coverage for -1, 86399999, and
86400000. See the Avro specification:
https://avro.apache.org/docs/1.11.2/specification/#time-millisecond-precision
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
