[20240822] XTDB Internals 4 - Relation

[metadata.clj]​

metadata 에 TableMetadata 는 Relation 을 인자로 받고 있다. 그리고 이것을 실행하는 ->table-metadata 함수가 relation 을 처음 만드는 것 같다. (metadata도 나중에 자세히 알아보자.)

(defn ->table-metadata ^xtdb.metadata.ITableMetadata [^IBufferPool buffer-pool ^Path file-path]
  (util/with-close-on-catch [buf (.getBuffer buffer-pool file-path)]
    (util/with-open [loader (Relation/loader buf)]
      (util/with-close-on-catch [rel (Relation. (.getAllocator (.getReferenceManager buf))
						(.getSchema loader))]
	(let [nodes-vec (.get rel "nodes")]
	  (.loadBatch loader 0 rel)
	  (let [rdr (.getOldRelReader rel)
		^IVectorReader metadata-reader (-> (.readerForName rdr "nodes")
						   (.legReader "leaf"))
		{:keys [col-names page-idx-cache]} (->table-metadata-idxs metadata-reader)
		temporal-col-types-rdr (some-> (.structKeyReader metadata-reader "columns")
					       (.listElementReader)
					       (.structKeyReader "types")
					       (.structKeyReader temporal-col-type-leg-name))

		min-rdr (some-> temporal-col-types-rdr (.structKeyReader "min"))
		max-rdr (some-> temporal-col-types-rdr (.structKeyReader "max"))]
	    (->TableMetadata (ArrowHashTrie. nodes-vec) rel buf metadata-reader col-names page-idx-cache (AtomicInteger. 1)
			     min-rdr max-rdr)))))))

->table-metadata 는 버퍼풀과 파일패스를 받아서 TableMetadata 를 만든다.

1. 버퍼 풀에서 파일에 대한 버퍼를 가져옵니다.
2. Relation 로더를 생성하고 첫 번째 배치 데이터를 로드합니다.
3. 메타데이터 정보(컬럼 이름, 페이지 인덱스 캐시 등)를 추출합니다.
4. 시간 관련 컬럼 타입에 대한 리더를 설정합니다.
5. 추출한 정보를 바탕으로 TableMetadata 인스턴스를 생성하여 반환합니다.

여기서 우리가 볼 것은 Relation 로더를 생성해서 데이터를 로드하는 부분이다.

@JvmStatic
fun loader(buf: ArrowBuf): Loader {
  buf.referenceManager.retain()
  return BufferLoader(buf, readFooter(buf))
}

    private class BufferLoader(
	private val buf: ArrowBuf,
	footer: ArrowFooter
    ) : Loader() {
	override val schema: Schema = footer.schema

	inner class Batch(private val idx: Int, private val block: ArrowBlock): Loader.Batch {

	    override fun load(rel: Relation) {
		val prefixSize =
		    if (buf.getInt(block.offset) == MessageSerializer.IPC_CONTINUATION_TOKEN) 8L else 4L

		val metadataBuf = buf.nioBuffer(block.offset + prefixSize, block.metadataLength - prefixSize.toInt())

		val bodyBuf = buf.slice(block.offset + block.metadataLength, block.bodyLength)
		    .also { it.referenceManager.retain() }

		val msg = Message.getRootAsMessage(metadataBuf.asReadOnlyBuffer())
		val recordBatchFB = RecordBatch().also { msg.header(it) }

		(MessageSerializer.deserializeRecordBatch(recordBatchFB, bodyBuf)
		    ?: error("Failed to deserialize record batch $idx, offset ${block.offset}"))

		    .use { batch -> rel.load(batch) }
	    }
	}

	override val batches = footer.recordBatches.mapIndexed(::Batch)

	override fun close() = buf.close()
    }

위 load 함수로 몇가지 구조를 유추할 수 있다.

1. 프리픽스 (Prefix)

크기: 4바이트 또는 8바이트 목적: 레코드 배치의 시작을 나타냄

1. 메타데이터 섹션

위치: 프리픽스 직후 크기: block.metadataLength - prefixSize로 결정됨 내용: 레코드 배치에 대한 직렬화된 메타데이터 접근 방법: buf.nioBuffer() 사용

1. 본문 섹션

위치: 메타데이터 섹션 이후 크기: block.bodyLength로 결정됨 내용: 실제 레코드 배치 데이터 접근 방법: buf.slice() 사용

1. 전체 구조

[프리픽스][메타데이터][본문]

여기서 나의 눈에 띄는 것은 schema 가 footer에 존재한다는 것이다. readFooter 함수가 궁금하다.