Implement Spark-compatible `from_csv`

未关闭
#293 0 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看

维护者通常 1 天内回复

还没有人认领这个 Issue。

评估

难度
5/5
预计耗时
一周以上
新手友好度
22/100
Issue 类型
功能
描述清晰度
基本清楚
活跃度
活跃
技术栈
rust, sql
领域
backend

调研方向

Start by reading crates/spark/src/function/csv/ (exists but empty) and the existing from_json implementation for the PERMISSIVE/FAILFAST pattern, then port Spark's csvExpressions.scala, UnivocityParser.scala and CSVExprUtils.scala semantics. Build a hand-written tokenizer mimicking univocity since arrow-csv does not fit, and compare against Comet's native/spark-expr/src/csv_funcs/to_csv.rs for option handling. Done means crates/sqllogictest/test_files/csv/from_csv.slt covers the listed edge cases and passes, with #[user_doc] documentation. This is a multi-week design-heavy port, not a starter task.

由索引模型根据 Issue 内容生成。

描述

enhancement

Part of #272 and apache/datafusion-comet#7300.

What the function does

from_csv(csvStr, schema[, options]) parses one CSV record into a STRUCT with the given schema. schema must be
a foldable string: a DDL string ('a INT, b DOUBLE') or schema_of_csv(...). A non-string schema fails with
INVALID_SCHEMA.NON_STRING_LITERAL, a non-struct type ('Array<int>') fails with INVALID_SCHEMA.NON_STRUCT_TYPE.
options must be a map(...) of string literals (INVALID_OPTIONS.NON_MAP_FUNCTION /
INVALID_OPTIONS.NON_STRING_TYPE). Option keys are case-insensitive. Field types can be any atomic type
(numeric, decimal, boolean, string, binary, date, timestamp, timestamp_ntz, intervals; VARIANT from 4.1, TIME
from 4.2). ARRAY, MAP and STRUCT fields fail at execution with UNSUPPORTED_DATATYPE.

The return type is schema.asNullable and the result is NULL only when the input is NULL. Since 3.0.0. The
function is time-zone aware (spark.sql.session.timeZone, or a timeZone option).

Parsing uses the univocity CsvParser configured from CSVOptions, wrapped in FailureSafeParser. Only
PERMISSIVE (default) and FAILFAST are allowed. DROPMALFORMED fails with PARSE_MODE_UNSUPPORTED.
The options that matter, with defaults: sep/delimiter (,, may be several characters, with \t-style escapes),
quote ("), escape (\), charToEscapeQuoteEscaping (unset), comment (disabled), nullValue (""),
emptyValue (""), nanValue (NaN), positiveInf (Inf), negativeInf (-Inf), ignoreLeadingWhiteSpace
and ignoreTrailingWhiteSpace (both false on read), mode, columnNameOfCorruptRecord
(spark.sql.columnNameOfCorruptRecord, default _corrupt_record), dateFormat, timestampFormat,
timestampNTZFormat, timeFormat (4.2), locale (en-US), enableDateTimeParsingFallback, maxColumns
(20480), maxCharsPerColumn (-1), unescapedQuoteHandling (STOP_AT_DELIMITER). lineSep is overridden with
\uFFFF, so an embedded newline is part of the field.

Examples
SELECT from_csv('1, 0.8', 'a INT, b DOUBLE');
-- {"a":1,"b":0.8}
SELECT from_csv('1, 3.14', 'a INT, f FLOAT');
-- {"a":1,"f":3.14}
SELECT from_csv('26/08/2015', 'time Timestamp', map('timestampFormat', 'dd/MM/yyyy'));
-- {"time":2015-08-26 00:00:00}
SELECT from_csv('1,abc', schema_of_csv('1,abc'));
-- {"_c0":1,"_c1":"abc"}
SELECT from_csv('14-30-45.123456', 'time TIME(6)', map('timeFormat', 'HH-mm-ss.SSSSSS'));   -- 4.2
-- {"time":14:30:45.123456}
SELECT from_csv('25:00:00', 'time TIME');                                                  -- 4.2
-- {"time":null}
SELECT from_csv(null, 'time TIME');
-- NULL

The examples above come from the ExpressionDescription and csv-functions.sql.out. The cases below come from
CsvExpressionsSuite and CsvFunctionsSuite (DataFrame API, shown here as SQL):

SELECT from_csv('---', 'a DOUBLE');                                           -- {"a":null}
SELECT from_csv(' ', 'a INT');                                                -- {"a":null}
SELECT from_csv('1', 'a INT, b INT');                                         -- {"a":1,"b":null}
SELECT from_csv('1,,"foo"', 'a BIGINT, b STRING, c STRING');                  -- {"a":1,"b":null,"c":"foo"}
SELECT from_csv('"#""', 'str STRING', map('escape', '#'));                     -- str is a single `"` character
SELECT from_csv('# This line is commented', 'str STRING', map('comment', '#')); -- {"str":null}
SELECT from_csv(' a   ', 'str STRING', map('ignoreLeadingWhiteSpace', 'true')); -- {"str":"a   "}
SELECT from_csv(' a   ', 'str STRING', map('ignoreTrailingWhiteSpace', 'true')); -- {"str":" a"}
SELECT from_csv('-', 'str STRING', map('nullValue', '-'));                    -- {"str":null}
SELECT from_csv('#', 'f FLOAT', map('nanValue', '#'));                        -- {"f":NaN}
SELECT from_csv('a,b,c,d', 'a STRING, b STRING, _corrupt_record STRING');     -- _corrupt_record = 'a,b,c,d' (a, b = 'a', 'b' by code reading)
SELECT from_csv('0,2013-111-11 12:13:14', 'a INT, b DATE, _unparsed STRING',
  map('columnNameOfCorruptRecord', '_unparsed'));                             -- {"a":0,"b":null,"_unparsed":"0,2013-111-11 12:13:14"} (timeParserPolicy CORRECTED)
SELECT from_csv('2021-08-12T15:16:23.000+11:00', 'a TIMESTAMP_NTZ');          -- {"a":null}
Edge cases to test
  • NULL input gives NULL. Any non-NULL input gives a non-NULL struct, even when every field is NULL.
  • PERMISSIVE partial results: a field that fails to convert becomes NULL and the other fields keep their values.
    A token count that differs from the schema marks the record malformed, but the tokens present are still
    converted (missing ones are NULL, extra ones are ignored). The corrupt-record column (must be a nullable STRING,
    otherwise INVALID_CORRUPT_RECORD_TYPE) gets the raw input only when the record is malformed.
  • FAILFAST fails with MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION. For input " and schema
    a INT, b INT, _unparsed STRING, badRecord is [null,null,"]. For 1,\u0001\u0000\u0001234 with a INT, b INT
    it is [1,null].
  • nullValue defaults to the empty string, so an empty field and a quoted empty field ("", read as emptyValue,
    also "") are both NULL even for STRING fields (by code reading of nullSafeDatum).
  • Whitespace is not trimmed by default. Numeric conversion uses the Java parsers: Integer.parseInt and
    Long.parseLong reject ' 2', but Float.parseFloat and Double.parseDouble trim, which is why ' 3.14' works
    above. Java float syntax also accepts 1d, 1f, hex floats and Infinity.
  • BOOLEAN uses Scala toBoolean (case-insensitive true/false only). DECIMAL with the default en-US locale
    strips every , before new BigDecimal(...), so a quoted "1,000" parses as 1000. Other locales use
    DecimalFormat. Values that do not fit the precision are NULL.
  • Date and timestamp: without dateFormat / timestampFormat the parse falls back to the lenient
    stringToDate / stringToTimestamp (cast rules). With a pattern set, no fallback unless
    enableDateTimeParsingFallback or spark.sql.legacy.csv.enableDateTimeParsingFallback is true.
    spark.sql.legacy.timeParserPolicy changes results: for 2013-111-11 12:13:14 as DATE, CORRECTED gives NULL,
    LEGACY gives 2022-03-11, EXCEPTION fails with INCONSISTENT_BEHAVIOR_CROSS_VERSION.PARSE_DATETIME_BY_NEW_PARSER.
    A string with an offset (Z) ignores the session time zone. locale affects month names (dd MMM yyyy HH:mm).
  • Options: a quote / escape / comment longer than one character fails with
    OPTION_VALUE_EXCEEDS_ONE_CHARACTER. An empty one means \u0000. Delimiter errors are INVALID_DELIMITER_VALUE.*
    (EMPTY_STRING, SINGLE_BACKSLASH, UNSUPPORTED_SPECIAL_CHARACTER, DELIMITER_LONGER_THAN_EXPECTED).
    maxCharsPerColumn = 2 on 12345 fails with a message containing
    Length of parsed input (3) exceeds the maximum number of characters defined in your parser settings (2).
  • Univocity quirks: an embedded newline stays in the field, because lineSep is forced to \uFFFF. Input
    1,2\n2 with schema schema_of_csv('1,2\n2') (_c0 INT, _c1 STRING) gives {1, "2\n2"} (SPARK-38955 test).
    Unbalanced quotes follow unescapedQuoteHandling (default STOP_AT_DELIMITER). Invalid UTF-8 input gives a
    struct of NULLs (from_csv bad UTF-8 test).
  • Scalar and column input; non-default spark.sql.session.timeZone; bad UTF-8; duplicate field names in the schema.
Differences across Spark versions
  • 3.4 / 3.5: spark.sql.legacy.timeParserPolicy defaults to EXCEPTION, so some malformed dates raise
    INCONSISTENT_BEHAVIOR_CROSS_VERSION.PARSE_DATETIME_BY_NEW_PARSER instead of giving NULL. 4.0 changes the default
    to CORRECTED.
  • 4.0: input accepts collated strings (StringTypeWithCollation, including trim collations). Evaluation moved into
    CsvToStructsEvaluator with codegen, with no change in results.
  • 4.1: VARIANT field type and the singleVariantColumn option.
  • 4.2: TIME(p) field type and the timeFormat option.
Implementation notes
  • Spark: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/csvExpressions.scala
    (CsvToStructs), expressions/csv/CsvExpressionEvalUtils.scala (CsvToStructsEvaluator),
    catalyst/csv/CSVOptions.scala, catalyst/csv/UnivocityParser.scala, catalyst/csv/CSVExprUtils.scala
    (delimiter parsing), catalyst/util/FailureSafeParser.scala.
  • DataFusion core has no from_csv. The arrow-csv reader is a file reader with different quoting, null and type
    rules, so a hand-written tokenizer that mimics univocity's settings is likely needed.
  • to_csv is tracked in apache/datafusion-spark#138 (port from Comet). Comet's native to_csv
    (native/spark-expr/src/csv_funcs/to_csv.rs, csv_write_options.rs) already models delimiter, quote,
    escape, nullValue, quoteAll and the whitespace flags. A shared CsvOptions struct (read and write defaults
    differ: whitespace flags are true on write, emptyValue is "" quoted on write) should live in
    crates/spark/src/function/csv/, which exists but is empty today.
  • Reuse the Spark date/timestamp parsing helpers in datafusion-spark if present, and keep from_json's
    PERMISSIVE / FAILFAST handling consistent if both are implemented.
  • Tests: add crates/sqllogictest/test_files/csv/from_csv.slt; document with #[user_doc].

This issue was generated by an LLM (Claude) from the Apache Spark source code and tests. Examples come from Spark's ExpressionDescription examples, test suites and .sql.out golden files, but please verify them against Spark before relying on them.

主要语言
Rust
星标
7
派生
15
平均合并
16 小时 43 分钟
30 天内合并 PR
26

环境准备

  • 没有 Dockerfile 或 Docker Compose 文件
  • 没有 Pull Request 模板
  • 阅读贡献指南

从这里开始

  1. 先读完整个 Issue,再读项目的贡献指南。
  2. 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
  3. Fork 仓库,在一个分支上完成修改。
  4. 提交 Pull Request,并在描述里引用这个 Issue 编号。

apache/datafusion-spark 的其他 Issue

查看 apache/datafusion-spark 的全部 Issue

相似的 Issue

更多 Rust Issue

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。