diff --git a/CHANGELOG.md b/CHANGELOG.md index f9d2e43c..09121e4c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,33 @@ true until the next version shipped. ### Fixed +- `pgcolumnar.import_arrow` reads an absent Arrow schema field as that field's + own default, so a float16 file is no longer accepted for a `float8` column + (#861). + + A RecordBatch buffer carries no type, so the Schema message is the only thing + that says what the bytes mean. The check for a `float8` target read + `FloatingPoint.precision` with a fallback of DOUBLE. `Schema.fbs` declares + `enum Precision : short { HALF, SINGLE, DOUBLE }` with no explicit default, so + an omitted precision means HALF, and a FlatBuffers writer omits any field that + equals its default: pyarrow writes float16 with no precision field at all. + Such a file passed the check and its 2-byte values were then read with the + target's 8-byte width. The fallback is now HALF, which is what the format + says, and the arm requires DOUBLE. + + `test/arrow_import.sh` gained the arms that hold the whole per-kind parameter + block, not just this one line. Each imports a real pyarrow file that shares + its target's Arrow type tag and differs in one parameter (uint64 for `bigint`, + timestamp[ms] for `timestamp`, date64 for `date`, decimal128(10,2) for + `numeric(20,4)`, and six more), and asserts SQLSTATE 42804 exactly. A file + that gets past the schema check and is caught later by a buffer bound reports + XX001 instead, so an arm asserting only "it failed" passes on the broken + build. Nine matching files are imported as the control, so a build that + refused every foreign file could not pass either. Disabling the parameter + block left the suite green before; it now reddens 13 checks. + +### Fixed + - A shebang and the execute bit go together, and every directory that documents a command is swept (#856). Two things were left over from #852. diff --git a/src/columnar_arrow.c b/src/columnar_arrow.c index cfa01631..aca77dc3 100644 --- a/src/columnar_arrow.c +++ b/src/columnar_arrow.c @@ -1212,6 +1212,11 @@ fbr_i32(const uint8 *b, uint32 len, uint32 pos) { return (int32) fbr_u32(b, len, pos); } +static int16 +fbr_i16(const uint8 *b, uint32 len, uint32 pos) +{ + return (int16) fbr_u16(b, len, pos); +} static int64 fbr_i64(const uint8 *b, uint32 len, uint32 pos) { @@ -1241,6 +1246,8 @@ pgc_fb_field(const uint8 *b, uint32 len, uint32 tab, int i) voff = fbr_u16(b, len, (uint32) vt + slot); if (voff == 0) return 0; + if ((uint64) tab + voff >= len) + IMPORT_CORRUPT("table field out of bounds"); return tab + voff; } @@ -1248,7 +1255,11 @@ pgc_fb_field(const uint8 *b, uint32 len, uint32 tab, int i) static uint32 pgc_fb_indirect(const uint8 *b, uint32 len, uint32 pos) { - return pos + fbr_u32(b, len, pos); + uint64 target = (uint64) pos + fbr_u32(b, len, pos); + + if (target > UINT32_MAX || target + 4 > len) + IMPORT_CORRUPT("offset target out of bounds"); + return (uint32) target; } /* Build a numeric input string for a 128-bit unscaled value at the given scale @@ -1321,6 +1332,7 @@ typedef struct ImpNode ArrowKind kind; Oid typid; int width; + int precision; int scale; int32 atttypmod; bool needsInput; @@ -1402,6 +1414,7 @@ imp_build_node(ImpNode *n, Oid typid, int32 typmod, bool *ok) return; } n->width = width; + n->precision = precision; n->scale = scale; n->needsInput = (n->kind == A_UTF8); if (n->needsInput) @@ -1414,6 +1427,217 @@ imp_build_node(ImpNode *n, Oid typid, int32 typmod, bool *ok) } } +/* + * Return element i of a FlatBuffers vector of table offsets. The vector length + * and every element slot are file-controlled, so validate the whole vector + * before using one of its offsets. + */ +static uint32 +imp_vector_table_at(const uint8 *b, uint32 len, uint32 vec, uint32 i) +{ + uint32 n = fbr_u32(b, len, vec); + uint64 slot; + + if ((uint64) n * 4 > (uint64) len - vec - 4) + IMPORT_CORRUPT("table vector runs past metadata"); + if (i >= n) + IMPORT_CORRUPT("table vector index out of range"); + slot = (uint64) vec + 4 + (uint64) i * 4; + return pgc_fb_indirect(b, len, (uint32) slot); +} + +static int32 +imp_i32_field(const uint8 *b, uint32 len, uint32 tab, int field, int32 def) +{ + uint32 pos = pgc_fb_field(b, len, tab, field); + + return pos ? fbr_i32(b, len, pos) : def; +} + +static int16 +imp_i16_field(const uint8 *b, uint32 len, uint32 tab, int field, int16 def) +{ + uint32 pos = pgc_fb_field(b, len, tab, field); + + return pos ? fbr_i16(b, len, pos) : def; +} + +static bool +imp_bool_field(const uint8 *b, uint32 len, uint32 tab, int field, bool def) +{ + uint32 pos = pgc_fb_field(b, len, tab, field); + + return pos ? fbr_u8(b, len, pos) != 0 : def; +} + +/* True when a FlatBuffers string field is present and non-empty. */ +static bool +imp_string_field_nonempty(const uint8 *b, uint32 len, uint32 tab, int field) +{ + uint32 pos = pgc_fb_field(b, len, tab, field); + uint32 str; + uint32 n; + + if (pos == 0) + return false; + str = pgc_fb_indirect(b, len, pos); + n = fbr_u32(b, len, str); + if ((uint64) n + 1 > (uint64) len - str - 4) + IMPORT_CORRUPT("string runs past metadata"); + return n > 0; +} + +/* + * Validate one Arrow schema Field against the PostgreSQL target node. + * + * RecordBatch buffers do not carry their types. Decoding them from the target + * tuple descriptor without checking the preceding Schema silently reinterprets + * equal-width values (for example float64 as int64), and mismatched nested + * layouts can assign every later buffer to the wrong column. Check the complete + * field tree before accepting any batch. + */ +static bool +imp_schema_field_matches(ImpNode *n, const uint8 *b, uint32 len, uint32 field) +{ + uint32 tagpos = pgc_fb_field(b, len, field, 2); + uint32 typepos = pgc_fb_field(b, len, field, 3); + uint32 type; + uint32 childrenpos = pgc_fb_field(b, len, field, 5); + uint32 children = 0; + uint32 nchildren = 0; + uint8 tag; + uint8 wanttag; + int i; + + if (tagpos == 0 || typepos == 0) + IMPORT_CORRUPT("schema field has no type"); + tag = fbr_u8(b, len, tagpos); + type = pgc_fb_indirect(b, len, typepos); + + switch (n->kind) + { + case A_INT16: + case A_INT32: + case A_INT64: + wanttag = ARROW_TYPE_Int; + break; + case A_FLOAT32: + case A_FLOAT64: + wanttag = ARROW_TYPE_FloatingPoint; + break; + case A_BINARY: + wanttag = ARROW_TYPE_Binary; + break; + case A_UTF8: + wanttag = ARROW_TYPE_Utf8; + break; + case A_BOOL: + wanttag = ARROW_TYPE_Bool; + break; + case A_DECIMAL128: + wanttag = ARROW_TYPE_Decimal; + break; + case A_DATE32: + wanttag = ARROW_TYPE_Date; + break; + case A_TIME64: + wanttag = ARROW_TYPE_Time; + break; + case A_TIMESTAMP: + case A_TIMESTAMPTZ: + wanttag = ARROW_TYPE_Timestamp; + break; + case A_LIST: + wanttag = ARROW_TYPE_List; + break; + case A_STRUCT: + wanttag = ARROW_TYPE_Struct; + break; + case A_UUID: + wanttag = ARROW_TYPE_FixedSizeBinary; + break; + default: + return false; + } + if (tag != wanttag) + return false; + + switch (n->kind) + { + case A_INT16: + case A_INT32: + case A_INT64: + if (imp_i32_field(b, len, type, 0, 0) != n->width * 8 || + false) + return false; + break; + case A_FLOAT32: + /* + * FloatingPoint { precision: Precision } and Precision is + * { HALF, SINGLE, DOUBLE } with no explicit default, so an omitted + * precision means HALF (0), not DOUBLE. pyarrow writes float16 with + * no precision field at all; a default of DOUBLE here passed such a + * file for a float8 column and 2-byte values were then read as + * 8-byte doubles. Pass the schema's own default and require the + * width the target needs. + */ + if (imp_i16_field(b, len, type, 0, 0) != 1) + return false; + break; + case A_FLOAT64: + if (imp_i16_field(b, len, type, 0, 0) != 2) + return false; + break; + case A_DATE32: + if (imp_i16_field(b, len, type, 0, 1) != 0) + return false; + break; + case A_TIME64: + if (imp_i16_field(b, len, type, 0, 1) != 2 || + imp_i32_field(b, len, type, 1, 32) != 64) + return false; + break; + case A_TIMESTAMP: + case A_TIMESTAMPTZ: + if (imp_i16_field(b, len, type, 0, 0) != 2) + return false; + if (imp_string_field_nonempty(b, len, type, 1) != + (n->kind == A_TIMESTAMPTZ)) + return false; + break; + case A_UUID: + if (imp_i32_field(b, len, type, 0, 0) != UUID_LEN) + return false; + break; + case A_DECIMAL128: + if (imp_i32_field(b, len, type, 0, 0) != n->precision || + imp_i32_field(b, len, type, 1, 0) != n->scale || + imp_i32_field(b, len, type, 2, 128) != 128) + return false; + break; + default: + break; + } + + if (childrenpos != 0) + { + children = pgc_fb_indirect(b, len, childrenpos); + nchildren = fbr_u32(b, len, children); + if ((uint64) nchildren * 4 > (uint64) len - children - 4) + IMPORT_CORRUPT("schema children vector runs past metadata"); + } + if (nchildren != (uint32) n->nchildren) + return false; + for (i = 0; i < n->nchildren; i++) + { + uint32 child = imp_vector_table_at(b, len, children, (uint32) i); + + if (!imp_schema_field_matches(&n->children[i], b, len, child)) + return false; + } + return true; +} + /* assign RecordBatch buffer indices to a node subtree, pre-order */ static void imp_assign_buffers(ImpNode *n, int *bufcur) @@ -1918,6 +2142,18 @@ pgcolumnar_import_arrow(PG_FUNCTION_ARGS) (errcode(ERRCODE_DATATYPE_MISMATCH), errmsg("Arrow file has %u columns, target table has %d", nfields, ncols))); + for (i = 0; i < ncols; i++) + { + uint32 field = imp_vector_table_at(meta, metaLen, fieldsVec, + (uint32) i); + + if (!imp_schema_field_matches(&tops[i], meta, metaLen, field)) + ereport(ERROR, + (errcode(ERRCODE_DATATYPE_MISMATCH), + errmsg("Arrow column %d does not match target column \"%s\"", + i + 1, + NameStr(TupleDescAttr(tupdesc, i)->attname)))); + } sawSchema = true; } else if (headerType == ARROW_MSG_RecordBatch) diff --git a/test/arrow_import.sh b/test/arrow_import.sh index c39a6a6e..4d539cff 100755 --- a/test/arrow_import.sh +++ b/test/arrow_import.sh @@ -144,6 +144,166 @@ with ipc.new_stream(pa.OSFile(sys.argv[1], 'wb'), t.schema) as w: PY psql_run "CREATE TABLE ri_d (c text) USING pgcolumnar;" expect_error "reject dictionary-encoded file" "SELECT pgcolumnar.import_arrow('ri_d', '$DICTF');" + + # RecordBatch buffers carry no type tags. The Schema must be checked before + # decoding, or equal-width values are silently reinterpreted (float64 bits as + # bigint here) and a nested mismatch shifts the buffer assignment of later + # columns. + MISMATCHF="$PGC_WORKDIR/type_mismatch.arrows" + python3 - "$MISMATCHF" <<'PY' +import sys, pyarrow as pa, pyarrow.ipc as ipc +t = pa.table({ + 'a': pa.array([1.5, 2.5], pa.float64()), + 'b': pa.array([[1, 2], [3]], pa.list_(pa.int32())), +}) +with ipc.new_stream(pa.OSFile(sys.argv[1], 'wb'), t.schema) as w: + w.write_table(t) +PY + psql_run "CREATE TABLE ri_type_mismatch (a bigint, b int[]) USING pgcolumnar;" + expect_error "reject equal-width scalar type mismatch" \ + "SELECT pgcolumnar.import_arrow('ri_type_mismatch', '$MISMATCHF');" + + psql_run "CREATE TABLE ri_nested_mismatch (a float8, b text) USING pgcolumnar;" + expect_error "reject nested schema mismatch" \ + "SELECT pgcolumnar.import_arrow('ri_nested_mismatch', '$MISMATCHF');" +fi + +# --------------------------------------------------------------------------- +# The Schema check must police EVERY per-kind type parameter, not just the +# type tag (issue #861). +# +# A RecordBatch buffer is untyped bytes. The tag switch alone accepts any +# Arrow type that shares a tag with the target: uint64 for bigint, float16 for +# float8, timestamp[ms] for timestamp, date64 for date. Those are read with +# the TARGET's width, sign and unit, so the rows arrive silently wrong (a +# timestamp[ms] file lands 1000x off) or a short buffer is read as a wide one. +# +# Each deny arm below shares its tag with its target and can only be rejected +# by the per-kind parameter block. Each asserts SQLSTATE 42804 +# (datatype_mismatch) exactly: "it failed" is not an assertion, because a +# typo, a missing table, a dead server and a downstream corruption check all +# fail too. A file that is caught only after the schema check reports XX001 +# (data_corrupted), which is a red here -- the mismatch has to be refused +# before any buffer is decoded. +# +# The accept arms are the control: without them a build that rejected every +# foreign file would pass every deny arm. +# --------------------------------------------------------------------------- +if [ "$have_pyarrow" = 1 ]; then + echo "-- Schema per-kind parameters are validated" + SVDIR="$PGC_WORKDIR/sv" + mkdir -p "$SVDIR" + python3 - "$SVDIR" <<'PY' +import sys, os, decimal +import pyarrow as pa, pyarrow.ipc as ipc + +d = sys.argv[1] + +def w(name, arr): + t = pa.table({'x': arr}) + with ipc.new_stream(pa.OSFile(os.path.join(d, name + '.arrows'), 'wb'), + t.schema) as o: + o.write_table(t) + +DAY = 86400000 + +# --- mismatched: shares the target's Arrow type tag, differs in a parameter -- +# FloatingPoint.precision has NO explicit default in Schema.fbs, so an omitted +# precision means HALF (0). pyarrow therefore writes NO precision field at all +# for float16 (verified on the bytes), which is exactly the case a default of +# DOUBLE would wave through. +w('f16', pa.array([1.5, 2.5, 3.5, 4.5], pa.float16())) +w('f16_zero', pa.array([], pa.float16())) +w('u64', pa.array([1, 2, 3, 4], pa.uint64())) +w('i32', pa.array([1, 2, 3, 4], pa.int32())) +w('ts_ms', pa.array([1, 2, 3, 4], pa.timestamp('ms'))) +w('ts_tz', pa.array([1, 2, 3, 4], pa.timestamp('us', tz='UTC'))) +w('ts_notz', pa.array([1, 2, 3, 4], pa.timestamp('us'))) +w('date64', pa.array([0, DAY, 2 * DAY, 3 * DAY], pa.date64())) +w('time32', pa.array([0, 1, 2, 3], pa.time32('ms'))) +w('fsb8', pa.array([b'12345678'] * 4, pa.binary(8))) +w('dec102', pa.array([decimal.Decimal('1.25')] * 4, pa.decimal128(10, 2))) + +# --- matching: the same tag WITH the parameters the target requires --------- +w('ok_f64', pa.array([1.5, 2.5, 3.5, 4.5], pa.float64())) +w('ok_f64_zero', pa.array([], pa.float64())) +w('ok_i64', pa.array([1, 2, 3, 4], pa.int64())) +w('ok_ts', pa.array([1, 2, 3, 4], pa.timestamp('us'))) +w('ok_tstz', pa.array([1, 2, 3, 4], pa.timestamp('us', tz='UTC'))) +w('ok_date32', pa.array([0, 1, 2, 3], pa.date32())) +w('ok_time64', pa.array([0, 1, 2, 3], pa.time64('us'))) +w('ok_fsb16', pa.array([b'0123456789abcdef'] * 4, pa.binary(16))) +w('ok_dec204', pa.array([decimal.Decimal('1.25')] * 4, pa.decimal128(20, 4))) +PY + + # Gate the premise: 20 fixture files, or the arms below prove nothing. + sv_n="$(ls "$SVDIR"/*.arrows 2>/dev/null | wc -l)" + if [ "$sv_n" -ne 20 ]; then + echo "FATAL: schema-parameter fixtures: got $sv_n files, want 20" >&2 + exit 1 + fi + + # One target table per Arrow type tag; every arm below reuses them, so a + # rejected import must leave them empty and an accepted one is counted. + psql_run "CREATE TABLE sv_f8 (x float8) USING pgcolumnar;" + psql_run "CREATE TABLE sv_i8 (x bigint) USING pgcolumnar;" + psql_run "CREATE TABLE sv_ts (x timestamp) USING pgcolumnar;" + psql_run "CREATE TABLE sv_tstz (x timestamptz) USING pgcolumnar;" + psql_run "CREATE TABLE sv_dt (x date) USING pgcolumnar;" + psql_run "CREATE TABLE sv_tm (x time) USING pgcolumnar;" + psql_run "CREATE TABLE sv_uu (x uuid) USING pgcolumnar;" + psql_run "CREATE TABLE sv_nm (x numeric(20,4)) USING pgcolumnar;" + + sv_deny() { # label, target table, fixture basename + check "reject $1 (42804)" \ + "$(sqlstate_or_hang "SELECT pgcolumnar.import_arrow('$2', '$SVDIR/$3.arrows')")" \ + "42804" + } + sv_accept() { # label, target table, fixture basename, expected rows + check "accept $1" \ + "$(q "SELECT pgcolumnar.import_arrow('$2', '$SVDIR/$3.arrows');" | tail -1)" \ + "$4" + } + + # FloatingPoint.precision. The zero-row file is the sharp one: nothing + # downstream can catch it, so the schema check is the only thing that can. + sv_deny "float16 into float8" sv_f8 f16 + sv_deny "empty float16 into float8" sv_f8 f16_zero + # Int.is_signed and Int.bitWidth. + sv_deny "uint64 into bigint" sv_i8 u64 + sv_deny "int32 into bigint" sv_i8 i32 + # Timestamp.unit and Timestamp.timezone (both directions). + sv_deny "timestamp[ms] into timestamp" sv_ts ts_ms + sv_deny "timestamp[us,UTC] into timestamp" sv_ts ts_tz + sv_deny "timestamp[us] into timestamptz" sv_tstz ts_notz + # Date.unit, Time.unit/bitWidth, FixedSizeBinary.byteWidth, Decimal p/s. + sv_deny "date64 into date" sv_dt date64 + sv_deny "time32[ms] into time" sv_tm time32 + sv_deny "fixed_size_binary(8) into uuid" sv_uu fsb8 + sv_deny "decimal128(10,2) into numeric(20,4)" sv_nm dec102 + + check "every rejected import left its target empty" \ + "$(q "SELECT (SELECT count(*) FROM sv_f8) + (SELECT count(*) FROM sv_i8) + + (SELECT count(*) FROM sv_ts) + (SELECT count(*) FROM sv_tstz) + + (SELECT count(*) FROM sv_dt) + (SELECT count(*) FROM sv_tm) + + (SELECT count(*) FROM sv_uu) + (SELECT count(*) FROM sv_nm);")" "0" + + # The controls: the same tags, with matching parameters, must go in. + sv_accept "float64 into float8" sv_f8 ok_f64 4 + sv_accept "empty float64 into float8" sv_f8 ok_f64_zero 0 + sv_accept "int64 into bigint" sv_i8 ok_i64 4 + sv_accept "timestamp[us] into timestamp" sv_ts ok_ts 4 + sv_accept "timestamp[us,UTC] into timestamptz" sv_tstz ok_tstz 4 + sv_accept "date32 into date" sv_dt ok_date32 4 + sv_accept "time64[us] into time" sv_tm ok_time64 4 + sv_accept "fixed_size_binary(16) into uuid" sv_uu ok_fsb16 4 + sv_accept "decimal128(20,4) into numeric(20,4)" sv_nm ok_dec204 4 + + # The accepted timestamp values must be the ones in the file, not a + # unit-scaled reading of them: 4 us past the epoch, never 4 ms. + check "accepted timestamp values are unscaled" \ + "$(q "SELECT string_agg(x::text, ',' ORDER BY x) FROM sv_ts;")" \ + "1970-01-01 00:00:00.000001,1970-01-01 00:00:00.000002,1970-01-01 00:00:00.000003,1970-01-01 00:00:00.000004" fi IXFILE="$PGC_WORKDIR/ix_roundtrip.arrows"