UCDP-AEC

Models and data for the UCDP Abstractive Event analysis Corpus paper
git clone https://esimon.eu/repos/UCDP-AEC.git
Log | Files | Refs | README | LICENSE

commit fe5801c926a6cf45e4fd42494cd1173827ca714e
parent 810312b31665c0ce683721c534c9eb4de112355c
Author: Étienne Simon <esimon@esimon.eu>
Date:   Mon, 11 May 2026 17:24:05 +0200

Prepare cache invalidation

Diffstat:
MREADME.md | 19+++++++++++--------
Maec/fuzzy_align.py | 2++
Maec/ids_to_documents.py | 110++++++++++++++++++++++++++++++++++++++++++++++++-------------------------------
Ahplt_align/zstd-index/Makefile | 3+++
Ahplt_align/zstd-index/line_offsets_of_seekable_zstd.c | 111+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Ahplt_align/zstd-index/shell.nix | 6++++++
Ahplt_align/zstd-index/zstd-seek.c | 655+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Ahplt_align/zstd-index/zstd-seek.h | 219+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
8 files changed, 1074 insertions(+), 51 deletions(-)

diff --git a/README.md b/README.md @@ -37,17 +37,20 @@ The `baselines` directory contains model code used in the experiments, some subd ## Citation -The proceedings are not published yet, but a preprint can be found [here](https://esimon.eu/CPSS2025/paper.pdf). +You can find our paper [in the ACL anthology](https://aclanthology.org/2025.konvens-2.8/). ``` @inproceedings{simon-etal-2025-abstractive, - title = {Abstractive Event Analysis of Armed Conflicts: Introducing the {UCDP-AEC} Dataset}, - author = {Simon, \'{E}tienne and Olsen, Helene B\o{}sei and Carre\~{n}o, Ram\'{o}n and Mishra, Rahul and Arefyev, Nikolay and Yilmaz, Mert Can and \O{}vrelid, Lilja and Velldal, Erik}, - year = {2025}, - month = sep, - booktitle = {Proceedings of the 5th Workshop on Computational Linguistics for the Political and Social Sciences}, - publisher = {Association for Computational Linguistics}, - address = {Hildesheim, Germany}, + title = "Abstractive Event Analysis of Armed Conflicts: Introducing the {UCDP}-{AEC} Dataset", + author = "Simon, {\'E}tienne and Olsen, Helene B{\o}sei and Carre{\~n}o, Ram{\'o}n and Mishra, Rahul and Arefyev, Nikolay and Yilmaz, Mert Can and {\O}vrelid, Lilja and Velldal, Erik", + editor = "Wartena, Christian and Heid, Ulrich", + booktitle = "Proceedings of the 21st Conference on Natural Language Processing (KONVENS 2025): Workshops", + month = sep, + year = "2025", + address = "Hannover, Germany", + publisher = "HsH Applied Academics", + url = "https://aclanthology.org/2025.konvens-2.8/", + pages = "104--119" } ``` diff --git a/aec/fuzzy_align.py b/aec/fuzzy_align.py @@ -23,6 +23,8 @@ class FuzzyAlign: def __call__(self, prediction): m = numpy.empty((len(prediction)+1, len(self.template)+1), dtype=numpy.int64) parents = numpy.zeros((len(prediction)+1, len(self.template)+1), dtype=numpy.int8) + parents[0, 1:] |= 2 + parents[1:, 0] |= 1 m[:, 0] = numpy.arange(len(prediction)+1) diff --git a/aec/ids_to_documents.py b/aec/ids_to_documents.py @@ -1,14 +1,17 @@ from typing import Any, Iterator import argparse +import bisect import collections +import io import json -import os +import operator import pathlib -import requests +import struct import sys import urllib.request import tqdm +import zstandard CACHE_URL: str = "https://recurrent.network/AEC/2025.cache" @@ -44,64 +47,85 @@ def process_jsonl_split(output_path: pathlib.Path, split_path: pathlib.Path, hpl print(json.dumps(sample), file=output_file) -def hplt_stream_lines(file: str) -> Iterator[str]: - import zstandard - response = requests.get(f"{HPLT_URL_PREFIX}/{file}", stream=True) - response.raise_for_status() - file_size = int(response.headers.get("content-length", 0)) - with tqdm.tqdm(desc=file, total=file_size, unit="B", unit_scale=True) as progress_bar: - decompressor = zstandard.ZstdDecompressor().decompressobj() - data_buffer: bytes = b"" - for chunk in response.iter_content(chunk_size=DOWNLOAD_CHUNK_SIZE): - progress_bar.update(len(chunk)) - data_buffer += decompressor.decompress(chunk) - if data_buffer: - lines: list[bytes] = data_buffer.split(b"\n") - data_buffer = lines[-1] - yield from map(bytes.decode, lines[:-1]) - if data_buffer: - yield data_buffer - - -def hplt_cacheless_download_lines(file: str, line_numbers: list[int]) -> dict[str, str]: +def hplt_build_index(url: str) -> list[tuple[int, int]]: + with urllib.request.urlopen(url) as file: + data: bytes = file.read() + + filename_length: int = data.find(b'\0') + header_length: int = filename_length + 5 + number_of_frames: int = struct.unpack("<I", data[filename_length+1:header_length])[0] + + cumulative_sum: tuple[int, int] = (0, 0) + index: list[tuple[int, int]] = [cumulative_sum] + for i in range(number_of_frames): + start: int = header_length + i*8 + newlines, compressed_size = struct.unpack("<2I", data[start:start+8]) + cumulative_sum = (cumulative_sum[0] + newlines, cumulative_sum[1] + compressed_size) + index.append(cumulative_sum) + return index + + +def hplt_get_line(url: str, index: list[tuple[str, str]], line_number: int) -> str: + right = bisect.bisect_right(index, line_number, key=operator.itemgetter(0)) + left = right - 1 + while left > 0 and index[left][0] == line_number: + left -= 1 + newlines_offset = index[left][0] + + start = index[left][1] + end = index[right][1] + + with urllib.request.urlopen( + urllib.request.Request(url, + headers={"Range": f"bytes={start}-{end-1}"})) as file: + data: bytes = file.read() + assert(len(data) == end - start) + + decompressor = zstandard.ZstdDecompressor() + reader = decompressor.stream_reader(io.BytesIO(data), read_across_frames=True) + return reader.readall().split(b'\n')[line_number - newlines_offset].decode() + + +def hplt_download_lines(filename: str, line_numbers: list[int]) -> dict[str, str]: + indexable_path: str = f"{HPLT_URL_PREFIX}/{filename[:-4]}.1M.zst" + index_path: str = f"{indexable_path}.zindex" + + index: list[tuple[int, int]] = hplt_build_index(index_path) + data: dict[str, str] = {} - lines_to_select: set[int] = set(line_numbers) line_number: int - line: str - for line_number, line in enumerate(hplt_stream_lines(file), start=1): - if line_number in lines_to_select: - document: dict[str, str] = json.loads(line) - data[f"{file}:{line_number}"] = document["text"] - lines_to_select.remove(line_number) - if not lines_to_select: - break - if lines_to_select: - raise RuntimeError(f"Invalid document id {file}:{lines_to_select}") + for line_number in set(line_numbers): + line: str = hplt_get_line(indexable_path, index, line_number - 1) + sample: dict[str, Any] = json.loads(line) + data[f"{filename}:{line_number}"] = sample["text"] return data -def hplt_cacheless_data(hplt_ids: list[str]) -> dict[str, str]: +def get_hplt_data(hplt_ids: list[str]) -> dict[str, str]: files: dict[str, list[int]] = collections.defaultdict(list) for hplt_id in hplt_ids: file, line_number = hplt_id.split(":") files[file].append(int(line_number)) data: dict[str, str] = {} - for file, line_numbers in files.items(): - data.update(hplt_cacheless_download_lines(file, line_numbers)) + for file, line_numbers in tqdm.tqdm(files.items(), desc="Downloading from HPLT"): + data.update(hplt_download_lines(file, line_numbers)) return data -def hplt_cached_data(hplt_ids: list[str]) -> dict[str, str]: +def get_hplt_cache(hplt_ids: list[str]) -> dict[str, str]: + cache: dict[str, str] = {} try: - cache: dict[str, str] = {} with urllib.request.urlopen(CACHE_URL) as cache_file: for line in cache_file: cached_document: dict[str, str] = json.loads(line) cache[cached_document["id"]] = cached_document["text"] - return cache - except: - print("\n===========================================================================================\nERROR: You need to update the UCDP-AEC repository, the current version has been deprecated.\n===========================================================================================", file=sys.stderr) - raise + except urllib.error.HTTPError as error: + if error.code == 404: + print("ERROR: The cache data was invalidated, try updating the repository or using the cacheless downloader (--no-cache).", file=sys.stderr) + sys.exit(1) + else: + raise + return cache def main(output_path: pathlib.Path, input_path: pathlib.Path, cacheless: bool, jsonl_only: bool) -> None: @@ -123,7 +147,7 @@ def main(output_path: pathlib.Path, input_path: pathlib.Path, cacheless: bool, j output_path.mkdir(parents=True) print("Downloading data…", flush=True, end=" ") - hplt_data: dict[str, str] = hplt_cacheless_data(hplt_ids) if cacheless else hplt_cached_data(hplt_ids) + hplt_data: dict[str, str] = get_hplt_data(hplt_ids) if cacheless else get_hplt_cache(hplt_ids) print("done") print("Transforming dataset…", flush=True, end=" ") diff --git a/hplt_align/zstd-index/Makefile b/hplt_align/zstd-index/Makefile @@ -0,0 +1,3 @@ +CC ?= gcc +all: + ${CC} -O3 -DNDEBUG ${CFLAGS} -lzstd ${LDFLAGS} line_offsets_of_seekable_zstd.c zstd-seek.c -o line_offsets_of_seekable_zstd diff --git a/hplt_align/zstd-index/line_offsets_of_seekable_zstd.c b/hplt_align/zstd-index/line_offsets_of_seekable_zstd.c @@ -0,0 +1,111 @@ +#include <assert.h> +#include <stdbool.h> +#include <stdint.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> + +#include "zstd-seek.h" + + +#ifndef NREAD +#define NREAD 100 +#endif + + +bool isLittleEndian(){ + volatile int x = 1; + return *(char*)(&x) == 1; +} + +void writeLE32(char **output, uint32_t data){ + if(isLittleEndian()){ + memcpy(*output, &data, sizeof(data)); + } else { + uint32_t swap = ((data & 0xFF000000) >> 24) | + ((data & 0x00FF0000) >> 8) | + ((data & 0x0000FF00) << 8) | + ((data & 0x000000FF) << 24); + memcpy(*output, &swap, sizeof(swap)); + } + *output += sizeof(uint32_t); +} + +int main(int argc, char **argv){ + if(argc != 2){ + fprintf(stderr, "format: %s path/to/seekable.file.zstd\n", argv[0]); + return EXIT_FAILURE; + } + + ZSTDSeek_Context *sctx = ZSTDSeek_createFromFile(argv[1]); + if(!sctx){ + fprintf(stderr, "Can't open zstd file.\n"); + return EXIT_FAILURE; + } + + ZSTDSeek_JumpTable *jt = ZSTDSeek_getJumpTableOfContext(sctx); + if(!jt){ + fprintf(stderr, "Zstd file not seekable.\n"); + return EXIT_FAILURE; + } + + size_t max_buffer = 0; + uint64_t last_uncompressed = 0; + for(ZSTDSeek_JumpTableRecord *record = jt->records; record != jt->records + jt->length; ++record){ + size_t candidate = record->uncompressedPos - last_uncompressed; + if(max_buffer < candidate) + max_buffer = candidate; + last_uncompressed = record->uncompressedPos; + } + + size_t output_size = strlen(argv[1]) + 1 + sizeof(uint32_t)*(1+2*(jt->length-1)); + char *output_start = malloc(output_size); + char *output_end = output_start + output_size; + char *output = output_start; + + strcpy(output, argv[1]); + output += strlen(argv[1]) + 1; + writeLE32(&output, jt->length-1); + + char *buffer = malloc(max_buffer * NREAD); + for(ZSTDSeek_JumpTableRecord *brecord = jt->records; brecord < jt->records + jt->length - 1; brecord += NREAD){ + assert(ZSTDSeek_tell(sctx) == brecord->uncompressedPos); + assert(output < output_end); + ZSTDSeek_JumpTableRecord *end = brecord+NREAD<jt->records+jt->length-1?brecord+NREAD:jt->records+jt->length-1; + + size_t read_size = 0; + for(ZSTDSeek_JumpTableRecord *record = brecord; record < end; ++record) + read_size += record[1].uncompressedPos - record->uncompressedPos; + size_t read = ZSTDSeek_read(buffer, read_size, sctx); + + size_t buffer_offset = 0; + for(ZSTDSeek_JumpTableRecord *record = brecord; record < end; ++record){ + uint32_t compressed_size = record[1].compressedPos - record->compressedPos; + uint32_t uncompressed_size = record[1].uncompressedPos - record->uncompressedPos; + + uint32_t newlines = 0; + for(size_t i = buffer_offset; i < buffer_offset+uncompressed_size; ++i) + if(buffer[i] == '\n') + ++newlines; + writeLE32(&output, newlines); + writeLE32(&output, compressed_size); + + buffer_offset += uncompressed_size; + } + } + + assert(output == output_end); + + char *output_name = malloc(strlen(argv[1]) + sizeof(".zindex")); + strcpy(output_name, argv[1]); + strcat(output_name, ".zindex"); + + FILE *output_file = fopen(output_name, "wb"); + size_t r = fwrite(output_start, output_size, 1, output_file); + fclose(output_file); + + free(output_name); + free(buffer); + ZSTDSeek_free(sctx); + return EXIT_SUCCESS; +} diff --git a/hplt_align/zstd-index/shell.nix b/hplt_align/zstd-index/shell.nix @@ -0,0 +1,6 @@ +{ pkgs ? import <nixpkgs> { }, }: +pkgs.mkShell { + packages = with pkgs; [ + zstd + ]; +} diff --git a/hplt_align/zstd-index/zstd-seek.c b/hplt_align/zstd-index/zstd-seek.c @@ -0,0 +1,655 @@ +/* ****************************************************************** + * libzstd-seek + * Copyright (c) 2020, Martinelli Marco + * + * You can contact the author at : + * - Source repository : https://github.com/martinellimarco/libzstd-seek + * + * This source code is licensed under the GPLv3 (found in the LICENSE + * file in the root directory of this source tree). +****************************************************************** */ + +#include <fcntl.h> +#include <stdlib.h> +#include <stdint.h> +#include <stdbool.h> +#include <sys/stat.h> +#include <string.h> +#include "zstd-seek.h" + +#ifdef _WIN32 +#include "windows-mmap.h" +#else +#include <unistd.h> +#include <sys/mman.h> +#endif + +typedef struct { + size_t compressedOffset; //how may bytes to skip from the beginning of the compressed stream (skip to target frame) + size_t uncompressedOffset; //how many bytes skip from the beginning of the uncompressed frame (move inside target frame) + ZSTDSeek_JumpTableRecord jtr; //copy of the jump table record that was used to calculate this jump coordinate +} ZSTDSeek_JumpCoordinate; + +struct ZSTDSeek_Context_s{ + ZSTD_DCtx* dctx; + + void *buff; //the start of the buffer with the zstd frame(s) + size_t size; //the length of buff + + size_t lastFrameCompressedSize; //the size of the last frame processed by read + + size_t currentUncompressedPos; //the current position in the uncompressed file, returned by tell + size_t currentCompressedPos; //the position in the compressed file, returned by compressedTell + + ZSTDSeek_JumpTable* jt; + int jumpTableFullyInitialized; + + ZSTDSeek_JumpCoordinate jc; + + size_t tmpOutBuffSize; + uint8_t* tmpOutBuff; + size_t tmpOutBuffPos; //the position where we read so far in the tmpOutBuff. if tmpOutBuffPos < output.pos we have data left in this buffer to read before we move on to uncompress more data + + int mmap_fd; //the file descriptor of the memory map, used only if the context is created with ZSTDSeek_createFromFile or ZSTDSeek_createFromFileWithoutJumpTable + int close_fd; //1 if we own the mmap_fd and we are responsible of closing it, 0 otherwise + + uint8_t* inBuff; //it's a pointer to something inside buff + ZSTD_inBuffer input; + ZSTD_outBuffer output; +}; + +/* Jump Table API */ + +ZSTDSeek_JumpTable* ZSTDSeek_getJumpTableOfContext(ZSTDSeek_Context *sctx){ + if(!sctx){ + return NULL; + } + return sctx->jt; +} + +ZSTDSeek_JumpTable* ZSTDSeek_newJumpTable(){ + ZSTDSeek_JumpTable *jt = malloc(sizeof(ZSTDSeek_JumpTable)); + + jt->records = malloc(sizeof(ZSTDSeek_JumpTableRecord)); + jt->length = 0; + jt->capacity = 1; + + return jt; +} + +void ZSTDSeek_freeJumpTable(ZSTDSeek_JumpTable* jt){ + if(!jt){ + DEBUG("Invalid argument"); + return; + } + free(jt->records); + free(jt); +} + +void ZSTDSeek_addJumpTableRecord(ZSTDSeek_JumpTable* jt, size_t compressedPos, size_t uncompressedPos){ + if(!jt){ + DEBUG("Invalid argument"); + return; + } + + if(jt->length == jt->capacity){ + if(jt->capacity == UINT64_MAX){ + DEBUG("Jump Table: Maximum capacity reached\n"); + return; + }else if(jt->capacity < (UINT64_MAX>>1)){ + jt->capacity *= 2; + }else{ + jt->capacity = UINT64_MAX; + } + jt->records = realloc(jt->records, jt->capacity*sizeof(ZSTDSeek_JumpTableRecord)); + } + + jt->records[jt->length++] = (ZSTDSeek_JumpTableRecord){ + compressedPos, + uncompressedPos + }; +} + +int ZSTDSeek_initializeJumpTable(ZSTDSeek_Context *sctx){ + return ZSTDSeek_initializeJumpTableUpUntilPos(sctx, SIZE_MAX); +} + +bool ZSTDSeek_isLittleEndian(){ + volatile int x = 1; + return *(char*)(&x) == 1; +} + +uint32_t ZSTDSeek_fromLE32(uint32_t data){ + if(ZSTDSeek_isLittleEndian()){ + return data; + }else{ + uint32_t swap = ((data & 0xFF000000) >> 24) | + ((data & 0x00FF0000) >> 8) | + ((data & 0x0000FF00) << 8) | + ((data & 0x000000FF) << 24); + return swap; + } +} + +int ZSTDSeek_initializeJumpTableUpUntilPos(ZSTDSeek_Context *sctx, size_t upUntilPos){ + if(!sctx){ + DEBUG("ZSTDSeek_Context is NULL\n"); + return -1; + } + + void *buff = sctx->buff; + size_t size = sctx->size; + + uint8_t *footer = (uint8_t *)buff + (size - ZSTD_SEEK_TABLE_FOOTER_SIZE); + uint32_t magicnumber = ZSTDSeek_fromLE32(*((uint32_t *)(footer + 5))); + + if(magicnumber == ZSTD_SEEKABLE_MAGICNUMBER){ + DEBUG("Seektable detected\n"); + uint8_t sfd = *((uint8_t*)(footer + 4)); + uint8_t checksumFlag = sfd >> 7; + + /* check reserved bits */ + if((sfd >> 2) & 0x1f){ + DEBUG("Last frame checksumFlag= %x: Bits 3-7 should be zero. Ignoring malformed seektable.\n",(uint32_t)sfd); + }else{ + uint32_t const numFrames = ZSTDSeek_fromLE32(*((uint32_t *)footer)); + uint32_t const sizePerEntry = 8 + (checksumFlag ? 4 : 0); + uint32_t const tableSize = sizePerEntry * numFrames; + uint32_t const frameSize = tableSize + ZSTD_SEEK_TABLE_FOOTER_SIZE + ZSTD_SKIPPABLE_HEADER_SIZE; + + uint8_t *frame = (uint8_t *)buff + (size - frameSize); + uint32_t skippableHeader = ZSTDSeek_fromLE32(*((uint32_t *)frame)); + if(skippableHeader != (ZSTD_MAGIC_SKIPPABLE_START|0xE)){ + DEBUG("Last frame Header = %u does not match magic number %u. Ignoring malformed seektable.\n", skippableHeader, (ZSTD_MAGIC_SKIPPABLE_START|0xE)); + }else{ + uint32_t _frameSize = ZSTDSeek_fromLE32(*((uint32_t *)(frame + 4))); + if(_frameSize + ZSTD_SKIPPABLE_HEADER_SIZE != frameSize){ + DEBUG("Last frame size = %u does not match expected size = %u. Ignoring malformed seektable.\n", _frameSize + ZSTD_SKIPPABLE_HEADER_SIZE, frameSize); + }else{ + uint8_t *table = frame + ZSTD_SKIPPABLE_HEADER_SIZE; + size_t cOffset = 0; + size_t dOffset = 0; + for(uint32_t i = 0; i < numFrames; i++){ + ZSTDSeek_addJumpTableRecord(sctx->jt, cOffset, dOffset); + cOffset += ZSTDSeek_fromLE32(*((uint32_t *)(table + (i * sizePerEntry)))); + dOffset += ZSTDSeek_fromLE32(*((uint32_t *)(table + (i * sizePerEntry) + 4))); + } + ZSTDSeek_addJumpTableRecord(sctx->jt, cOffset, dOffset); + + sctx->jumpTableFullyInitialized = 1; + return 0; + } + } + } + } + + size_t frameCompressedSize; + size_t compressedPos = 0; + size_t uncompressedPos = 0; + + if(sctx->jt->length > 0){ + compressedPos = sctx->jt->records[sctx->jt->length-1].compressedPos; + uncompressedPos = sctx->jt->records[sctx->jt->length-1].uncompressedPos; + } + + buff = (uint8_t *)sctx->buff + compressedPos; + + sctx->jumpTableFullyInitialized = 1; + + while ((frameCompressedSize = ZSTD_findFrameCompressedSize(buff, size))>0 && !ZSTD_isError(frameCompressedSize)) { + uint32_t const magic = ZSTDSeek_fromLE32(*((uint32_t *)buff)); + if((magic & ZSTD_MAGIC_SKIPPABLE_MASK) == ZSTD_MAGIC_SKIPPABLE_START){ + compressedPos += frameCompressedSize; + buff = (uint8_t *)buff + frameCompressedSize; + continue; + } + + if(sctx->jt->length == 0 || sctx->jt->records[sctx->jt->length-1].uncompressedPos < uncompressedPos){ + ZSTDSeek_addJumpTableRecord(sctx->jt, compressedPos, uncompressedPos); + } + + size_t frameContentSize = ZSTD_getFrameContentSize(buff, size); + if(ZSTD_isError(frameContentSize)){//true if the uncompressed size is not known + frameContentSize = 0; + + ZSTD_DCtx *dctx = ZSTD_createDCtx(); + size_t const buffOutSize = ZSTD_DStreamOutSize(); + void* const buffOut = malloc(buffOutSize); + size_t lastRet = 0; + void *buffIn = buff; + + ZSTD_inBuffer input = { buffIn, frameCompressedSize, 0 }; + while (input.pos < input.size) { + ZSTD_outBuffer output = { buffOut, buffOutSize, 0 }; + lastRet = ZSTD_decompressStream(dctx, &output , &input); + if(ZSTD_isError(lastRet)){ + DEBUG("Error decompressing: %s\n", ZSTD_getErrorName(lastRet)); + ZSTD_freeDCtx(dctx); + free(buffOut); + return -1; + } + frameContentSize += output.pos; + } + ZSTD_freeDCtx(dctx); + free(buffOut); + + if (lastRet != 0) { + DEBUG("Unexpected EOF. Is the file truncated?\n"); + return -1; + } + } + + compressedPos += frameCompressedSize; + uncompressedPos += frameContentSize; + buff = (uint8_t *)buff + frameCompressedSize; + + if(uncompressedPos >= upUntilPos){ + sctx->jumpTableFullyInitialized = 0; + break; + } + } + if(sctx->jt->length > 0){ + if(sctx->jt->records[sctx->jt->length-1].uncompressedPos < uncompressedPos){ + ZSTDSeek_addJumpTableRecord(sctx->jt, compressedPos, uncompressedPos); + } + return 0; + }else{ //0 frames found + DEBUG("No frames\n"); + return -1; + } +} + +int ZSTDSeek_jumpTableIsInitialized(ZSTDSeek_Context *sctx){ + return sctx->jumpTableFullyInitialized; +} + +ZSTDSeek_JumpCoordinate ZSTDSeek_getJumpCoordinate(ZSTDSeek_Context *sctx, size_t uncompressedPos) { + if(!sctx->jumpTableFullyInitialized && (sctx->jt->length == 0 || sctx->jt->records[sctx->jt->length-1].uncompressedPos <= uncompressedPos)){ + ZSTDSeek_initializeJumpTableUpUntilPos(sctx, uncompressedPos); + } + + //search for the greater value of m where sctx->jt->records[m].uncompressedPos <= uncompressedPos + uint64_t l = 0; + uint64_t r = sctx->jt->length - 1; + while(l <= r){ + uint64_t m = (l+r)/2; + if(sctx->jt->records[m].uncompressedPos > uncompressedPos){ + r = m-1; + }else if((m+1) < sctx->jt->length && sctx->jt->records[m+1].uncompressedPos <= uncompressedPos){ + l = m+1; + }else{ + return (ZSTDSeek_JumpCoordinate){sctx->jt->records[m].compressedPos, uncompressedPos - sctx->jt->records[m].uncompressedPos, sctx->jt->records[m]}; + } + } +/* + //old linear search + for(uint64_t i = sctx->jt->length - 1; i >= 0; i--){ + if(sctx->jt->records[i].uncompressedPos <= uncompressedPos){ + return (ZSTDSeek_JumpCoordinate){sctx->jt->records[i].compressedPos, uncompressedPos - sctx->jt->records[i].uncompressedPos, sctx->jt->records[i]}; + } + } +*/ + return (ZSTDSeek_JumpCoordinate){0, uncompressedPos, (ZSTDSeek_JumpTableRecord){0, 0}}; +} + +/* Seek API */ + +ZSTDSeek_Context* ZSTDSeek_createFromFileWithoutJumpTable(const char* file){ + struct stat st; + stat(file, &st); + + int fd = open(file, O_RDONLY, 0); + if(fd < 0){ + DEBUG("Unable to open '%s'\n", file); + return NULL; + } + + void* const buff = mmap(NULL, st.st_size, PROT_READ, MAP_PRIVATE, fd, 0); + if(buff == MAP_FAILED){ + DEBUG("Unable to mmap '%s'\n", file); + return NULL; + } + + ZSTDSeek_Context *sctx = ZSTDSeek_createWithoutJumpTable(buff, st.st_size); + if(sctx){ + sctx->mmap_fd = fd; + sctx->close_fd = 1; + return sctx; + }else{ + munmap(buff, st.st_size); + close(fd); + return NULL; + } +} + +ZSTDSeek_Context* ZSTDSeek_createFromFile(const char* file){ + ZSTDSeek_Context* sctx = ZSTDSeek_createFromFileWithoutJumpTable(file); + if(sctx && ZSTDSeek_initializeJumpTable(sctx)!=0){ + DEBUG("Can't initialize the jump table\n"); + ZSTDSeek_free(sctx); + return NULL; + } + return sctx; +} + +ZSTDSeek_Context* ZSTDSeek_createFromFileDescriptorWithoutJumpTable(int fd){ + size_t size = lseek(fd,0L,SEEK_END); + + void* const buff = mmap(NULL, size, PROT_READ, MAP_PRIVATE, fd, 0); + if(buff == MAP_FAILED){ + DEBUG("Unable to mmap file descriptor %d\n", fd); + return NULL; + } + + ZSTDSeek_Context *sctx = ZSTDSeek_createWithoutJumpTable(buff, size); + if(sctx){ + sctx->mmap_fd = fd; + sctx->close_fd = 0; + return sctx; + }else{ + munmap(buff, size); + close(fd); + return NULL; + } +} + +ZSTDSeek_Context* ZSTDSeek_createFromFileDescriptor(int fd){ + ZSTDSeek_Context* sctx = ZSTDSeek_createFromFileDescriptorWithoutJumpTable(fd); + if(sctx && ZSTDSeek_initializeJumpTable(sctx)!=0){ + DEBUG("Can't initialize the jump table\n"); + ZSTDSeek_free(sctx); + return NULL; + } + return sctx; +} + +ZSTDSeek_Context* ZSTDSeek_createWithoutJumpTable(void *buff, size_t size){ + ZSTD_DCtx *dctx = ZSTD_createDCtx(); + + ZSTDSeek_Context* sctx = malloc(sizeof(ZSTDSeek_Context)); + + sctx->dctx = dctx; + + sctx->buff = buff; + sctx->size = size; + + sctx->inBuff = (uint8_t*)buff; + + sctx->currentUncompressedPos = 0; + sctx->currentCompressedPos = 0; + + sctx->lastFrameCompressedSize = 0; + + sctx->jc = (ZSTDSeek_JumpCoordinate){0,0}; + + sctx->tmpOutBuffSize = ZSTD_DStreamOutSize(); + sctx->tmpOutBuff = (uint8_t*)malloc(sctx->tmpOutBuffSize); + sctx->tmpOutBuffPos = 0; + + sctx->mmap_fd = -1; + sctx->close_fd = 0; + + sctx->input = (ZSTD_inBuffer){sctx->inBuff, 0, 0}; + sctx->output = (ZSTD_outBuffer){sctx->tmpOutBuff, 0, 0}; + + sctx->jt = ZSTDSeek_newJumpTable(); + sctx->jumpTableFullyInitialized = 0; + + //test if the buffer starts with a valid frame + if(ZSTD_isError(ZSTD_findFrameCompressedSize(sctx->buff, sctx->size))){ + DEBUG("Invalid format\n"); + ZSTDSeek_free(sctx); + return NULL; + } + + return sctx; +} + +ZSTDSeek_Context* ZSTDSeek_create(void *buff, size_t size){ + ZSTDSeek_Context* sctx = ZSTDSeek_createWithoutJumpTable(buff, size); + if(sctx && ZSTDSeek_initializeJumpTable(sctx)!=0){ + DEBUG("Can't initialize the jump table\n"); + ZSTDSeek_free(sctx); + return NULL; + } + return sctx; +} + +size_t ZSTDSeek_read(void *outBuff, size_t outBuffSize, ZSTDSeek_Context *sctx){ + if(!sctx){ + DEBUG("ZSTDSeek_Context is NULL\n"); + return 0; + } + + ZSTDSeek_JumpCoordinate localJc = ZSTDSeek_getJumpCoordinate(sctx, sctx->currentUncompressedPos); //trigger the generation of a jump table record, if needed + sctx->currentCompressedPos = localJc.jtr.compressedPos; + + size_t maxReadable = ZSTDSeek_lastKnownUncompressedFileSize(sctx) - sctx->currentUncompressedPos; + size_t toRead = maxReadable < outBuffSize ? maxReadable : outBuffSize; + size_t shouldRead = toRead; + + if(sctx->tmpOutBuffPos < sctx->output.pos){ + if(sctx->jc.uncompressedOffset > sctx->output.pos){ + sctx->jc.uncompressedOffset -= sctx->output.pos; + }else{ + size_t maxCopy = (sctx->output.pos - sctx->tmpOutBuffPos) - sctx->jc.uncompressedOffset; + size_t toCopy = maxCopy < toRead ? maxCopy : toRead; + + memcpy(outBuff, sctx->tmpOutBuff+sctx->tmpOutBuffPos+sctx->jc.uncompressedOffset, toCopy); + toRead -= toCopy; + outBuff = (uint8_t *)outBuff + toCopy; + sctx->currentUncompressedPos += toCopy; + sctx->tmpOutBuffPos += toCopy + sctx->jc.uncompressedOffset; + sctx->jc.uncompressedOffset = 0; + } + } + + while (toRead > 0 && ((sctx->input.pos < sctx->input.size) || (sctx->lastFrameCompressedSize = ZSTD_findFrameCompressedSize(sctx->inBuff, sctx->size)) > 0)){ + if(sctx->input.pos == sctx->input.size){ + sctx->input = (ZSTD_inBuffer){sctx->inBuff, sctx->lastFrameCompressedSize, 0}; + } + + while (sctx->input.pos < sctx->input.size) { + sctx->output = (ZSTD_outBuffer){ sctx->tmpOutBuff, sctx->tmpOutBuffSize, 0 }; + sctx->tmpOutBuffPos = 0; + size_t const ret = ZSTD_decompressStream(sctx->dctx, &sctx->output , &sctx->input); + + if(ZSTD_isError(ret)){ + DEBUG("Error decompressing: %s\n", ZSTD_getErrorName(ret)); + return ZSTDSEEK_ERR_READ; + } + + sctx->currentCompressedPos += sctx->input.pos; + + if(sctx->jc.uncompressedOffset > sctx->output.pos){ + sctx->jc.uncompressedOffset -= sctx->output.pos; + }else{ + size_t maxCopy = (sctx->output.pos - sctx->tmpOutBuffPos) - sctx->jc.uncompressedOffset; + size_t toCopy = maxCopy < toRead ? maxCopy : toRead; + + memcpy(outBuff, sctx->tmpOutBuff+sctx->tmpOutBuffPos+sctx->jc.uncompressedOffset, toCopy); + toRead -= toCopy; + outBuff = (uint8_t *)outBuff + toCopy; + sctx->currentUncompressedPos += toCopy; + sctx->tmpOutBuffPos += toCopy + sctx->jc.uncompressedOffset; + sctx->jc.uncompressedOffset = 0; + } + + if(toRead == 0){ + break; + } + } + + if(sctx->input.pos == sctx->input.size){ //end of frame + sctx->inBuff+=sctx->lastFrameCompressedSize; + } + + if(toRead == 0){ + break; + } + } + + return shouldRead - toRead; +} + +int ZSTDSeek_seek(ZSTDSeek_Context *sctx, long offset, int origin){ + if(!sctx){ + DEBUG("ZSTDSeek_Context is NULL\n"); + return -1; + } + if(origin == SEEK_CUR){ + if(offset==0){ + return 0; + } + offset = (long)sctx->currentUncompressedPos + offset; + origin = SEEK_SET; + }else if(origin == SEEK_END){ + offset = (long)ZSTDSeek_uncompressedFileSize(sctx) + offset; + origin = SEEK_SET; + } + if(origin == SEEK_SET){ + if(offset < 0){ + DEBUG("Negative seek\n"); + return ZSTDSEEK_ERR_NEGATIVE_SEEK; + }else if(offset > 0){ + ZSTDSeek_getJumpCoordinate(sctx, sctx->currentUncompressedPos+offset); //trigger an update of the lastKnownUncompressedFileSize + if(offset > ZSTDSeek_lastKnownUncompressedFileSize(sctx)){ + DEBUG("Seek to a frame beyond the buffer length\n"); + return ZSTDSEEK_ERR_BEYOND_END_SEEK; + } + } + + if(offset == sctx->currentUncompressedPos){ //we are already there, do nothing + return 0; + } + + ZSTDSeek_JumpCoordinate new_jc = ZSTDSeek_getJumpCoordinate(sctx, offset); + + if(sctx->jc.compressedOffset != new_jc.compressedOffset || offset < sctx->currentUncompressedPos){ //reset + ZSTD_DCtx_reset(sctx->dctx, ZSTD_reset_session_only); + + sctx->jc = new_jc; + + sctx->inBuff = (uint8_t *)sctx->buff + sctx->jc.compressedOffset; //jump to the beginning of the frame.. + sctx->currentUncompressedPos = offset; //..and adjust the uncompressed position.. + sctx->currentCompressedPos = sctx->jc.compressedOffset; + sctx->tmpOutBuffPos = 0; //..and reset the position in the tmp buffer + sctx->input = (ZSTD_inBuffer){sctx->inBuff, 0, 0}; + sctx->output = (ZSTD_outBuffer){sctx->tmpOutBuff, 0, 0}; + + }else{ //move forward + size_t toSkipTotal = offset - sctx->currentUncompressedPos; + + size_t const buffOutSize = ZSTD_DStreamOutSize(); + void* const buffOut = malloc(buffOutSize); + + while(toSkipTotal>0){ + size_t toSkip = buffOutSize < toSkipTotal ? buffOutSize : toSkipTotal; + toSkipTotal -= ZSTDSeek_read(buffOut, toSkip, sctx); + } + + free(buffOut); + } + }else{ + DEBUG("Invalid origin\n"); + return -1; + } + + return 0; +} + +long ZSTDSeek_tell(ZSTDSeek_Context *sctx){ + if(!sctx){ + DEBUG("ZSTDSeek_Context is NULL\n"); + return -1; + } + + return sctx->currentUncompressedPos; +} + +long ZSTDSeek_compressedTell(ZSTDSeek_Context *sctx){ + if(!sctx){ + DEBUG("ZSTDSeek_Context is NULL\n"); + return -1; + } + + return sctx->currentCompressedPos; +} + +size_t ZSTDSeek_uncompressedFileSize(ZSTDSeek_Context *sctx){ + if(!sctx){ + DEBUG("ZSTDSeek_Context is NULL\n"); + return 0; + } + + ZSTDSeek_initializeJumpTable(sctx); + + return ZSTDSeek_lastKnownUncompressedFileSize(sctx); +} + +size_t ZSTDSeek_lastKnownUncompressedFileSize(ZSTDSeek_Context *sctx){ + if(!sctx){ + DEBUG("ZSTDSeek_Context is NULL\n"); + return 0; + } + + return sctx->jt->length > 0 ? sctx->jt->records[sctx->jt->length-1].uncompressedPos : 0; +} + +int ZSTDSeek_fileno(ZSTDSeek_Context *sctx){ + return sctx->mmap_fd; +} + +size_t ZSTDSeek_countFramesUpTo(ZSTDSeek_Context *sctx, size_t upTo){ + if(!sctx){ + DEBUG("ZSTDSeek_Context is NULL\n"); + return 0; + } + + size_t frameCompressedSize; + + void *buff = sctx->buff; + size_t size = sctx->size; + + size_t counter = 0; + + while ((frameCompressedSize = ZSTD_findFrameCompressedSize(buff, size))>0 && !ZSTD_isError(frameCompressedSize)) { + counter++; + buff = (uint8_t *)buff + frameCompressedSize; + if(counter >= upTo){ + return upTo; + } + } + + return counter; +} + +size_t ZSTDSeek_getNumberOfFrames(ZSTDSeek_Context *sctx){ + return ZSTDSeek_countFramesUpTo(sctx, SIZE_MAX); +} + +int ZSTDSeek_isMultiframe(ZSTDSeek_Context *sctx){ + return ZSTDSeek_countFramesUpTo(sctx, 2) > 1; +} + +void ZSTDSeek_free(ZSTDSeek_Context *sctx){ + if(!sctx){ + DEBUG("ZSTDSeek_Context is NULL\n"); + return; + } + + if(sctx->dctx){ + ZSTD_freeDCtx(sctx->dctx); + } + + ZSTDSeek_freeJumpTable(sctx->jt); + + if(sctx->mmap_fd>=0 && sctx->close_fd){ + munmap(sctx->buff, sctx->size); + close(sctx->mmap_fd); + } + + free(sctx->tmpOutBuff); + + free(sctx); +} diff --git a/hplt_align/zstd-index/zstd-seek.h b/hplt_align/zstd-index/zstd-seek.h @@ -0,0 +1,219 @@ +/* ****************************************************************** + * libzstd-seek + * Copyright (c) 2020, Martinelli Marco + * + * You can contact the author at : + * - Source repository : https://github.com/martinellimarco/libzstd-seek + * + * This source code is licensed under the GPLv3 (found in the LICENSE + * file in the root directory of this source tree). +****************************************************************** */ + +#ifndef _ZSTD_SEEK_ +#define _ZSTD_SEEK_ + +#if defined (__cplusplus) +extern "C" { +#endif + +#include <stdio.h> +#include <zstd.h> + +#ifndef _ZSTD_SEEK_DEBUG_ +#define _ZSTD_SEEK_DEBUG_ 0 +#endif + +#ifdef _WIN32 +#define DEBUG(...) +#else +#define DEBUG(...) \ +do { \ + if (_ZSTD_SEEK_DEBUG_){ \ + flockfile(stderr); \ + fprintf(stderr, "%s(): ", __func__); \ + fprintf(stderr, __VA_ARGS__); \ + funlockfile(stderr); \ + } \ +} while (0) +#endif + +/* Error constants */ +#define ZSTDSEEK_ERR_NEGATIVE_SEEK -1 +#define ZSTDSEEK_ERR_BEYOND_END_SEEK -2 +#define ZSTDSEEK_ERR_READ -3 + +/* Seekable format constants */ +#define ZSTD_SEEK_TABLE_FOOTER_SIZE 9 +#define ZSTD_SEEKABLE_MAGICNUMBER 0x8F92EAB1 +#define ZSTD_SKIPPABLE_HEADER_SIZE 8 + +/* Structs */ + +typedef struct{ + size_t compressedPos; //where this frame begin in the compressed stream + size_t uncompressedPos;//the equivalent position in the uncompressed stream +} ZSTDSeek_JumpTableRecord; + +typedef struct{ + ZSTDSeek_JumpTableRecord *records; + uint64_t length; + uint64_t capacity; +} ZSTDSeek_JumpTable; + +typedef struct ZSTDSeek_Context_s ZSTDSeek_Context; + +/* Jump Table API */ + +/* + * Returns a pointer to the jump table used by the given context, 0 if the context is not valid. + */ +ZSTDSeek_JumpTable* ZSTDSeek_getJumpTableOfContext(ZSTDSeek_Context *sctx); + +/* + * These are for advanced use. Don't use them unless you understand exactly what you are doing. + */ +ZSTDSeek_JumpTable* ZSTDSeek_newJumpTable(); +void ZSTDSeek_freeJumpTable(ZSTDSeek_JumpTable* jt); + +/* + * Add a new record to the jump table jt. It's a simple map between compressed and uncompressed positions. + * Uncompressed positions must be at the start of a frame. + * The last record is special. The compressedPos is the compressed file size and the uncompressedPos is the uncompressed file size. + */ +void ZSTDSeek_addJumpTableRecord(ZSTDSeek_JumpTable* jt, size_t compressedPos, size_t uncompressedPos); + +/* + * Parse the file and fill the jump table. Don't use in combination with addJumpTableRecord. + * You don't need to call this unless you used a create*WithoutJumpTable method to get the context. + */ +int ZSTDSeek_initializeJumpTable(ZSTDSeek_Context *sctx); + +/* + * Parse the file and fill the jump table up until a certain uncompressed position. Don't use in combination with addJumpTableRecord. + * You don't need to call this unless you used a create*WithoutJumpTable method to get the context. + */ +int ZSTDSeek_initializeJumpTableUpUntilPos(ZSTDSeek_Context *sctx, size_t upUntilPos); + +/* + * Return 1 if the jump table is fully initialized, 0 otherwise. + */ +int ZSTDSeek_jumpTableIsInitialized(ZSTDSeek_Context *sctx); + +/* Seek API */ + +/* + * Create a ZSTDSeek_Context from a file and initialize the jump table. + * file is the path to the file + * On success returns a valid ZSTDSeek_Context. + * Returns ZSTDSEEK_CREATE_OPEN_FAILED if the file can't be opened. + * Returns ZSTDSEEK_CREATE_MAP_FAILED if the file can't be memory mapped. + */ +ZSTDSeek_Context* ZSTDSeek_createFromFile(const char* file); + +/* + * Create a ZSTDSeek_Context from a file but does not initialze the jump table. + * You'll have to call ZSTDSeek_initializeJumpTable manually or add records with ZSTDSeek_addJumpTableRecord. + * file is the path to the file + * On success returns a valid ZSTDSeek_Context. + * Returns ZSTDSEEK_CREATE_OPEN_FAILED if the file can't be opened. + * Returns ZSTDSEEK_CREATE_MAP_FAILED if the file can't be memory mapped. + */ +ZSTDSeek_Context* ZSTDSeek_createFromFileWithoutJumpTable(const char* file); + +/* + * Create a ZSTDSeek_Context from a buffer and initialize the jump table. + * buff is a buffer with the whole zstd data, eg a memory mapped file. + * length is the length of buff. + * Returns 0 in case of failure. + */ +ZSTDSeek_Context* ZSTDSeek_create(void *buff, size_t size); + +/* + * Create a ZSTDSeek_Context from a buffer but does not initialze the jump table. + * You'll have to call ZSTDSeek_initializeJumpTable manually or add records with ZSTDSeek_addJumpTableRecord. + * buff is a buffer with the whole zstd data, eg a memory mapped file. + * length is the length of buff. + * Returns 0 in case of failure. + */ +ZSTDSeek_Context* ZSTDSeek_createWithoutJumpTable(void *buff, size_t size); + +/* + * Create a ZSTDSeek_Context from a file descriptor but does not initialze the jump table. + * You'll have to call ZSTDSeek_initializeJumpTable manually or add records with ZSTDSeek_addJumpTableRecord. + * USE WITH CAUTION. The file descriptor will be used with mmap. + * fd is the file descriptor + * On success returns a valid ZSTDSeek_Context. + * Returns ZSTDSEEK_CREATE_MAP_FAILED if the fd can't be memory mapped. + */ +ZSTDSeek_Context* ZSTDSeek_createFromFileDescriptorWithoutJumpTable(int fd); + +/* + * Create a ZSTDSeek_Context from a file descriptor and initialize the jump table. + * USE WITH CAUTION. The file descriptor will be used with mmap. + * fd is the file descriptor + * On success returns a valid ZSTDSeek_Context. + * Returns ZSTDSEEK_CREATE_MAP_FAILED if the fd can't be memory mapped. + */ +ZSTDSeek_Context* ZSTDSeek_createFromFileDescriptor(int fd); + +/* + * It reads outBuffSize bytes of uncompressed data from the sctx context buffer into outBuff. + * Returns the number of bytes read. + */ +size_t ZSTDSeek_read(void *outBuff, size_t outBuffSize, ZSTDSeek_Context *sctx); + +/* + * Like fseek. + * Origin can be SEEK_SET, SEEK_END or SEEK_CUR. + * Returns 0 if seek was successfull. + */ +int ZSTDSeek_seek(ZSTDSeek_Context *sctx, long offset, int origin); + +/* + * Like ftell. + * Returns the current position in the uncompressed file or -1 in case of failure. + */ +long ZSTDSeek_tell(ZSTDSeek_Context *sctx); + +/* + * Like ftell. + * Returns the current position in the compressed file or -1 in case of failure. + */ +long ZSTDSeek_compressedTell(ZSTDSeek_Context *sctx); + +/* + * Return the size of the uncompressed file. It will trigger the initialization of the full jump table. + */ +size_t ZSTDSeek_uncompressedFileSize(ZSTDSeek_Context *sctx); + +/* + * Return the last known size of the uncompressed file. It's the size of all the uncompressed frames discovered so far stored in the jump table. + */ +size_t ZSTDSeek_lastKnownUncompressedFileSize(ZSTDSeek_Context *sctx); + +/* + * Return the file descriptor associated with the memory mapped file if available, -1 otherwise. + * USE WITH CAUTION. IF YOU DON'T KNOW WHY YOU NEED IT YOU DON'T NEED IT. + */ +int ZSTDSeek_fileno(ZSTDSeek_Context *sctx); + +/* + * Returns the number of frames in the file. + */ +size_t ZSTDSeek_getNumberOfFrames(ZSTDSeek_Context *sctx); + +/* + * Returns 1 if there are more than one frame in the file. + */ +int ZSTDSeek_isMultiframe(ZSTDSeek_Context *sctx); + +/* + * Free the context. + */ +void ZSTDSeek_free(ZSTDSeek_Context *sctx); + +#if defined (__cplusplus) +} +#endif + +#endif