JingsongLi commented on code in PR #69:
URL: https://github.com/apache/paimon-mosaic/pull/69#discussion_r3835280248
##########
cli/src/main.rs:
##########
@@ -429,48 +530,815 @@ 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;
+// Hard ceiling on a single raw JSON record. Guards against a hostile input
+// where one record is arbitrarily large:
`Deserializer::into_iter::<Box<RawValue>>`
+// buffers the whole record before yielding, and
`normalize_json_decimal_record`
+// reallocates it into a fresh `String`. Enforced before normalization runs.
+const MAX_JSON_RECORD_BYTES: usize = 256 * 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();
+ if raw_bytes.len() > MAX_JSON_RECORD_BYTES {
Review Comment:
[P2] Enforce the record limit before allocation on every JSON path
The limit is checked only after serde_json has materialized the complete
Box<RawValue>, so a record larger than available memory can exhaust the process
before reaching this branch. In addition, inferred schemas and explicit schemas
without special-value validation take the Arrow reader path at lines 554-568
and never consult MAX_JSON_RECORD_BYTES. Please put the limit in a bounded
record reader shared by inference and decoding so it fires before either parser
allocates the complete record.
##########
cli/src/main.rs:
##########
@@ -429,48 +530,815 @@ 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;
+// Hard ceiling on a single raw JSON record. Guards against a hostile input
+// where one record is arbitrarily large:
`Deserializer::into_iter::<Box<RawValue>>`
+// buffers the whole record before yielding, and
`normalize_json_decimal_record`
+// reallocates it into a fresh `String`. Enforced before normalization runs.
+const MAX_JSON_RECORD_BYTES: usize = 256 * 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();
+ if raw_bytes.len() > MAX_JSON_RECORD_BYTES {
+ return Err(invalid_schema(format!(
+ "JSON record {record} is {} bytes, exceeds the {} byte limit",
+ raw_bytes.len(),
+ MAX_JSON_RECORD_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::Time32(TimeUnit::Millisecond) if !raw.get().starts_with('"')
=> {
+ validate_json_integer(raw, data_type, path, record, 0,
i128::from(MAX_TIME_MILLIS))?;
+ }
+ DataType::Time32(TimeUnit::Millisecond) => {
+ let value: String = serde_json::from_str(raw.get()).map_err(|_| {
+ invalid_schema(format!(
+ "JSON field '{}' at record {record} must be a valid
time-millis value",
+ fmt::safe(path)
+ ))
+ })?;
+ let parsed = Time32MillisecondType::parse(&value).ok_or_else(|| {
+ invalid_schema(format!(
+ "JSON field '{}' at record {record} must be a valid
time-millis value; got '{}'",
+ fmt::safe(path),
+ fmt::safe(&value)
+ ))
+ })?;
+ if !valid_time_millis(parsed) {
+ return Err(invalid_schema(format!(
+ "JSON field '{}' at record {record} is out of range for
{data_type}; got '{}'",
+ fmt::safe(path),
+ fmt::safe(&value)
+ )));
+ }
+ }
+ DataType::Int32 | DataType::Date32 if !raw.get().starts_with('"') => {
+ validate_json_integer(
+ raw,
+ data_type,
+ path,
+ record,
+ i128::from(i32::MIN),
+ i128::from(i32::MAX),
+ )?;
+ }
+ DataType::Int64 | DataType::Timestamp(_, _) if
!raw.get().starts_with('"') => {
+ validate_json_integer(
+ raw,
+ data_type,
+ path,
+ record,
+ i128::from(i64::MIN),
+ i128::from(i64::MAX),
+ )?;
+ }
+ DataType::Decimal128(precision, scale) => {
+ let raw_text = raw.get();
+ let value = if raw_text.starts_with('"') {
+ std::borrow::Cow::Owned(
+ serde_json::from_str::<String>(raw_text)
+ .map_err(|e| invalid_schema(format!("invalid JSON
decimal: {e}")))?,
+ )
+ } else {
+ std::borrow::Cow::Borrowed(raw_text)
+ };
+ parse_decimal_exact(&value, *precision, *scale).map_err(|e| {
+ invalid_schema(format!(
+ "cannot parse '{}' as {data_type} for JSON field '{}' at
record {record}: {e}",
+ fmt::safe(&value),
+ fmt::safe(path)
+ ))
+ })?;
+ }
+ DataType::List(field) => {
+ let values: Vec<&RawValue> = serde_json::from_str(raw.get())
+ .map_err(|e| invalid_schema(format!("invalid JSON array:
{e}")))?;
+ let child_path = format!("{path}[]");
+ for value in values {
+ validate_json_special_value(value, field, &child_path,
record)?;
+ }
+ }
+ DataType::Map(entries, _) => {
+ let DataType::Struct(fields) = entries.data_type() else {
+ return Ok(());
+ };
+ let Some(value_field) = fields.get(1) else {
+ return Ok(());
+ };
+ let child_path = format!("{path}{{}}");
+ let mut deserializer =
serde_json::Deserializer::from_str(raw.get());
+ JsonSpecialMapSeed {
+ value_field,
+ path: &child_path,
+ record,
+ }
+ .deserialize(&mut deserializer)
+ .map_err(|e| invalid_schema(format!("invalid JSON map: {e}")))?;
+ }
+ DataType::Struct(fields) => {
+ let values: std::collections::HashMap<String, &RawValue> =
+ serde_json::from_str(raw.get())
+ .map_err(|e| invalid_schema(format!("invalid JSON object:
{e}")))?;
+ for field in fields {
+ if let Some(value) = values.get(field.name()) {
+ let child_path = format!("{path}.{}", field.name());
+ validate_json_special_value(value, field, &child_path,
record)?;
+ }
+ }
+ }
+ DataType::Timestamp(_, None) if raw.get().starts_with('"') => {
+ let value: String = serde_json::from_str(raw.get())
+ .map_err(|e| invalid_schema(format!("invalid JSON timestamp:
{e}")))?;
+ if timestamp_has_explicit_timezone(&value) {
+ return Err(invalid_schema(format!(
+ "JSON field '{}' at record {record} must not include a
timezone for a local timestamp; got '{}'",
+ fmt::safe(path),
+ fmt::safe(&value)
+ )));
+ }
+ }
+ _ => {}
+ }
+ Ok(())
+}
+
+fn validate_avro_uuid(value: &str) -> Result<(), ()> {
+ let bytes = value.as_bytes();
+ if bytes.len() != 36 {
+ return Err(());
+ }
+ for (index, byte) in bytes.iter().enumerate() {
+ if matches!(index, 8 | 13 | 18 | 23) {
+ if *byte != b'-' {
+ return Err(());
+ }
+ } else if !byte.is_ascii_hexdigit() {
+ return Err(());
+ }
+ }
+ Ok(())
+}
+
+fn validate_json_integer(
+ raw: &RawValue,
+ data_type: &DataType,
+ path: &str,
+ record: usize,
+ min: i128,
+ max: i128,
+) -> std::io::Result<()> {
+ let value = raw.get();
+ let parsed = parse_decimal_exact(value, 38, 0).map_err(|_| {
+ invalid_schema(format!(
+ "JSON field '{}' at record {record} must be an integer for
{data_type}; got '{}'",
+ fmt::safe(path),
+ fmt::safe(value)
+ ))
+ })?;
+ if parsed < min || parsed > max {
+ return Err(invalid_schema(format!(
+ "JSON field '{}' at record {record} is out of range for
{data_type}; got '{}'",
+ fmt::safe(path),
+ fmt::safe(value)
+ )));
+ }
+ Ok(())
+}
+
+const MAX_TIME_MILLIS: i32 = 86_399_999;
+
+fn valid_time_millis(value: i32) -> bool {
+ (0..=MAX_TIME_MILLIS).contains(&value)
+}
+
+struct CsvConvertOptions {
+ delimiter: String,
+ escape: Option<String>,
+ quote: String,
+ no_header: bool,
+ header: Option<String>,
+ skip_lines: usize,
+}
+
+fn convert_csv(
+ inputs: &[PathBuf],
+ out: &Path,
+ schema: Option<&Path>,
+ required_fields: &[String],
+ options: CsvConvertOptions,
+ stats: Option<&str>,
+ overwrite: bool,
+) -> std::io::Result<()> {
+ if inputs.is_empty() {
+ return Err(std::io::Error::new(
+ std::io::ErrorKind::InvalidInput,
+ "CSV path is required",
+ ));
+ }
+ if schema.is_some() && !required_fields.is_empty() {
+ return Err(invalid_schema(
+ "--require applies only when the schema is inferred; set
nullability in the --schema file instead",
+ ));
+ }
+ ensure_can_write(out, overwrite)?;
+ let format = csv_format(&options)?;
+ let explicit_schema = schema.map(load_convert_schema).transpose()?;
+ let has_explicit_schema = explicit_schema.is_some();
+ let schema = match explicit_schema {
+ Some(schema) => schema,
+ None => {
+ let mut inferred: Option<Schema> = None;
+ for input in inputs {
+ let (reader, line_offset) = open_csv(input,
options.skip_lines)?;
+ let (schema, rows) = format
+ .infer_schema(reader, None)
Review Comment:
[P2] Reject oversized CSV width before schema inference
For inferred CSVs, format.infer_schema runs before open_csv_input applies
MAX_CSV_COLUMNS. A wide input can therefore make Arrow allocate names, types,
and Fields for every column before the rejection. The positional path also sets
layout.columns to zero for --no-header, so it bypasses the limit altogether.
Please validate the first header or data record width before inference and
carry the observed width through the positional layout.
--
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]
