读时模式

工作原理

写入方只存储字节和最少的分帧信息，例如文件格式、文档或日志行。读取方声明字段和类型，引擎在扫描时解析每条记录。在数据湖中，表定义把文件映射为列；在文档存储中，应用把文档映射为对象。无效记录可能被跳过、作为 null 返回，或使整个查询失败，取决于引擎和设置。

这与写时模式正好相反：数据库在插入时校验并转换数据。带类型列的关系表会立即拒绝格式错误的值；原始 Parquet 文件可能在读取方期望整数的地方包含字符串，而错误直到很久以后才在财务报表中暴露。

何时使用

对原始落地区、日志、事件归档和模式确实未知的探索性分析使用读时模式。它让你在决定如何建模之前保留数据，并支持在同一位置存放异构记录。在数据服务受监管报表或下游契约之前，应转向声明并校验的模式。

权衡

灵活性把工作推给每次查询和每个消费者。每次扫描都可能解析和转换数据，消耗 CPU，并可能产生不一致的结果。坏记录发现得晚，可能已经被看板或模型消费。治理也更困难，因为有效模式存在于查询中，而不是一条目录记录里。

示例与误解

团队把点击流 JSON 落入对象存储。一名分析人员把 event_time 定义为时间戳并成功查询；另一名分析人员用不同的时间格式读取同一批文件，却静默得到 null。解决办法是发布读取模式并在银层校验，而不是再写一条临时查询。

误解是不加约束地读取没有成本。它只是把成本转嫁给消费者，并让数据质量变成每个人各自的私事。

代码示例

export interface ValidationIssue {
  readonly path: string;
  readonly message: string;
}

export type ValidationResult =
  | { readonly valid: true }
  | { readonly valid: false; readonly issues: ValidationIssue[] };

type Schema = {
  readonly type?: "object" | "array" | "string" | "number" | "boolean" | "null";
  readonly required?: readonly string[];
  readonly properties?: Record<string, Schema>;
  readonly items?: Schema;
  readonly minLength?: number;
  readonly minimum?: number;
  readonly enum?: readonly unknown[];
};

function typeOf(value: unknown): string {
  if (value === null) return "null";
  if (Array.isArray(value)) return "array";
  return typeof value;
}

function childPath(path: string, segment: string | number): string {
  // 数组下标用方括号，属性用点号——贴近常见 JSON path 书写习惯
  if (typeof segment === "number") return path + "[" + segment + "]";
  return path === "$" ? "$." + segment : path + "." + segment;
}

export function validate(value: unknown, schema: Schema, path = "$"): ValidationResult {
  const issues: ValidationIssue[] = [];
  walk(value, schema, path, issues);
  return issues.length === 0 ? { valid: true } : { valid: false, issues };
}

function walk(value: unknown, schema: Schema, path: string, issues: ValidationIssue[]): void {
  if (schema.type && typeOf(value) !== schema.type) {
    issues.push({
      path,
      message: "expected " + schema.type + " but got " + typeOf(value),
    });
    return;
  }

  if (schema.enum && !schema.enum.includes(value)) {
    issues.push({ path, message: "value must be one of " + JSON.stringify(schema.enum) });
  }

  if (schema.type === "string" && typeof value === "string") {
    if (schema.minLength !== undefined && value.length < schema.minLength) {
      issues.push({ path, message: "string shorter than " + schema.minLength + " characters" });
    }
  }

  if (schema.type === "number" && typeof value === "number") {
    if (schema.minimum !== undefined && value < schema.minimum) {
      issues.push({ path, message: "number below minimum " + schema.minimum });
    }
  }

  if (schema.type === "object" && value && typeof value === "object" && !Array.isArray(value)) {
    const record = value as Record<string, unknown>;
    for (const key of schema.required ?? []) {
      if (!(key in record)) {
        issues.push({ path, message: "missing required property " + key });
      }
    }
    for (const [key, childSchema] of Object.entries(schema.properties ?? {})) {
      if (key in record) walk(record[key], childSchema, childPath(path, key), issues);
    }
  }

  if (schema.type === "array" && Array.isArray(value)) {
    if (schema.items) {
      value.forEach((item, index) => walk(item, schema.items as Schema, childPath(path, index), issues));
    }
  }
}

package main

import (
	"fmt"
	"math/bits"
)

type Bitset []uint64

func NewBitset(size int) Bitset {
	if size < 0 {
		panic("negative bitset size")
	}
	return make(Bitset, (size+63)/64)
}

func (b Bitset) Set(index int) {
	b[index/64] |= uint64(1) << (uint(index) % 64)
}

func (b Bitset) Clear(index int) {
	b[index/64] &^= uint64(1) << (uint(index) % 64)
}

func (b Bitset) Has(index int) bool {
	return b[index/64]&(uint64(1)<<(uint(index)%64)) != 0
}

func (b Bitset) Union(other Bitset) Bitset {
	size := len(b)
	if len(other) > size {
		size = len(other)
	}
	out := make(Bitset, size)
	for i := range out {
		var left, right uint64
		if i < len(b) {
			left = b[i]
		}
		if i < len(other) {
			right = other[i]
		}
		out[i] = left | right
	}
	return out
}

func (b Bitset) Intersect(other Bitset) Bitset {
	size := len(b)
	if len(other) < size {
		size = len(other)
	}
	out := make(Bitset, size)
	for i := range out {
		out[i] = b[i] & other[i]
	}
	return out
}

func (b Bitset) Count() int {
	total := 0
	for _, word := range b {
		total += bits.OnesCount64(word)
	}
	return total
}

func (b Bitset) Slice() []int {
	out := make([]int, 0, b.Count())
	for wordIndex, word := range b {
		for word != 0 {
			bit := bits.TrailingZeros64(word)
			out = append(out, wordIndex*64+bit)
			word &^= uint64(1) << uint(bit)
		}
	}
	return out
}

func main() {
	active := NewBitset(80)
	for _, id := range []int{1, 5, 64, 79} {
		active.Set(id)
	}
	blocked := NewBitset(80)
	blocked.Set(5)
	blocked.Set(9)

	fmt.Println("count:", active.Count())
	fmt.Println("has 64:", active.Has(64), "has 65:", active.Has(65))
	fmt.Println("members:", active.Slice())
	fmt.Println("union count:", active.Union(blocked).Count())
	fmt.Println("intersect:", active.Intersect(blocked).Slice())

	active.Clear(64)
	fmt.Println("after clear:", active.Count())
}