From eb1305c88e204bd04cb072f1bdafda515583d1b3 Mon Sep 17 00:00:00 2001 From: John Kerl Date: Sun, 31 Jul 2016 18:28:12 -0400 Subject: [PATCH] mlr head early-out feature --- c/lib/context.h | 1 + c/mapping/mapper_head.c | 38 ++++++++++++++++++++++++------ c/mapping/mapper_join.c | 8 ++++++- c/stream/stream.c | 4 +++- c/todo.txt | 2 ++ c/unit_test/test_rval_evaluators.c | 12 +++++----- 6 files changed, 50 insertions(+), 15 deletions(-) diff --git a/c/lib/context.h b/c/lib/context.h index 431a1fc8d..381b24297 100644 --- a/c/lib/context.h +++ b/c/lib/context.h @@ -8,6 +8,7 @@ typedef struct _context_t { long long fnr; int filenum; char* filename; + int force_eof; // e.g. mlr head } context_t; void context_init(context_t* pctx, char* first_file_name); diff --git a/c/mapping/mapper_head.c b/c/mapping/mapper_head.c index 75e154744..c9b9ec9e6 100644 --- a/c/mapping/mapper_head.c +++ b/c/mapping/mapper_head.c @@ -14,6 +14,7 @@ typedef struct _mapper_head_state_t { ap_state_t* pargp; slls_t* pgroup_by_field_names; unsigned long long head_count; + unsigned long long unkeyed_record_count; lhmslv_t* precord_lists_by_group; } mapper_head_state_t; @@ -21,7 +22,8 @@ static void mapper_head_usage(FILE* o, char* argv0, char* verb); static mapper_t* mapper_head_parse_cli(int* pargi, int argc, char** argv); static mapper_t* mapper_head_alloc(ap_state_t* pargp, slls_t* pgroup_by_field_names, unsigned long long head_count); static void mapper_head_free(mapper_t* pmapper); -static sllv_t* mapper_head_process(lrec_t* pinrec, context_t* pctx, void* pvstate); +static sllv_t* mapper_head_process_unkeyed(lrec_t* pinrec, context_t* pctx, void* pvstate); +static sllv_t* mapper_head_process_keyed(lrec_t* pinrec, context_t* pctx, void* pvstate); // ---------------------------------------------------------------- mapper_setup_t mapper_head_setup = { @@ -65,10 +67,13 @@ static mapper_t* mapper_head_alloc(ap_state_t* pargp, slls_t* pgroup_by_field_na pstate->pargp = pargp; pstate->pgroup_by_field_names = pgroup_by_field_names; pstate->head_count = head_count; + pstate->unkeyed_record_count = 0LL; pstate->precord_lists_by_group = lhmslv_alloc(); pmapper->pvstate = pstate; - pmapper->pprocess_func = mapper_head_process; + pmapper->pprocess_func = pgroup_by_field_names->length == 0 + ? mapper_head_process_unkeyed + : mapper_head_process_keyed; pmapper->pfree_func = mapper_head_free; return pmapper; @@ -90,20 +95,39 @@ static void mapper_head_free(mapper_t* pmapper) { } // ---------------------------------------------------------------- -static sllv_t* mapper_head_process(lrec_t* pinrec, context_t* pctx, void* pvstate) { +static sllv_t* mapper_head_process_unkeyed(lrec_t* pinrec, context_t* pctx, void* pvstate) { mapper_head_state_t* pstate = pvstate; if (pinrec != NULL) { - slls_t* pgroup_by_field_values = mlr_reference_selected_values_from_record(pinrec, pstate->pgroup_by_field_names); + pstate->unkeyed_record_count++; + if (pstate->unkeyed_record_count <= pstate->head_count) { + return sllv_single(pinrec); + } else { + pctx->force_eof = TRUE; + lrec_free(pinrec); + return NULL; + } + } else { + return sllv_single(NULL); + } +} + +// ---------------------------------------------------------------- +static sllv_t* mapper_head_process_keyed(lrec_t* pinrec, context_t* pctx, void* pvstate) { + mapper_head_state_t* pstate = pvstate; + if (pinrec != NULL) { + slls_t* pgroup_by_field_values = mlr_reference_selected_values_from_record(pinrec, + pstate->pgroup_by_field_names); if (pgroup_by_field_values == NULL) { lrec_free(pinrec); return NULL; } else { - unsigned long long* pcount_for_group = lhmslv_get(pstate->precord_lists_by_group, pgroup_by_field_values); + unsigned long long* pcount_for_group = lhmslv_get(pstate->precord_lists_by_group, + pgroup_by_field_values); if (pcount_for_group == NULL) { pcount_for_group = mlr_malloc_or_die(sizeof(unsigned long long)); *pcount_for_group = 0LL; - lhmslv_put(pstate->precord_lists_by_group, slls_copy(pgroup_by_field_values), pcount_for_group, - FREE_ENTRY_KEY); + lhmslv_put(pstate->precord_lists_by_group, slls_copy(pgroup_by_field_values), + pcount_for_group, FREE_ENTRY_KEY); } slls_free(pgroup_by_field_values); (*pcount_for_group)++; diff --git a/c/mapping/mapper_join.c b/c/mapping/mapper_join.c index 975239fce..99c94de6b 100644 --- a/c/mapping/mapper_join.c +++ b/c/mapping/mapper_join.c @@ -554,7 +554,13 @@ static void ingest_left_file(mapper_join_state_t* pstate) { pstate->popts->left_file_name); plrec_reader->psof_func(plrec_reader->pvstate, pvhandle); - context_t ctx = { .nr = 0, .fnr = 0, .filenum = 1, .filename = pstate->popts->left_file_name }; + context_t ctx = { + .nr = 0, + .fnr = 0, + .filenum = 1, + .filename = pstate->popts->left_file_name, + .force_eof = FALSE + }; context_t* pctx = &ctx; pstate->pleft_buckets_by_join_field_values = lhmslv_alloc(); diff --git a/c/stream/stream.c b/c/stream/stream.c index fe76bc78c..e8592aa39 100644 --- a/c/stream/stream.c +++ b/c/stream/stream.c @@ -35,7 +35,7 @@ int do_stream_chained(char* prepipe, slls_t* filenames, lrec_reader_t* plrec_rea exit(1); } - context_t ctx = { .nr = 0, .fnr = 0, .filenum = 0, .filename = NULL }; + context_t ctx = { .nr = 0, .fnr = 0, .filenum = 0, .filename = NULL, .force_eof = FALSE }; int ok = 1; if (filenames == NULL) { // No input at all @@ -83,6 +83,8 @@ static int do_file_chained(char* prepipe, char* filename, context_t* pctx, lrec_t* pinrec = plrec_reader->pprocess_func(plrec_reader->pvstate, pvhandle, pctx); if (pinrec == NULL) break; + if (pctx->force_eof == TRUE) // e.g. mlr head + break; pctx->nr++; pctx->fnr++; diff --git a/c/todo.txt b/c/todo.txt index 2ebcccf16..56e181e3c 100644 --- a/c/todo.txt +++ b/c/todo.txt @@ -21,6 +21,8 @@ TOP OF LIST: ! shift-by-1 (shift-by-n) at cookbook. also at mapper step. +!! mapper head UT inside various then-chains at various positions + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - COMMON: * mlr -k needs >/>> examples incl. stdout/stderr/file. also some subset at mlr put -h. diff --git a/c/unit_test/test_rval_evaluators.c b/c/unit_test/test_rval_evaluators.c index 6fdf1b839..84c3fbae9 100644 --- a/c/unit_test/test_rval_evaluators.c +++ b/c/unit_test/test_rval_evaluators.c @@ -12,7 +12,7 @@ int assertions_failed = 0; static char * test_caps() { printf("\n"); printf("-- TEST_RVAL_EVALUATORS test_caps ENTER\n"); - context_t ctx = {.nr = 888, .fnr = 999, .filenum = 123, .filename = "filename-goes-here"}; + context_t ctx = {.nr = 888, .fnr = 999, .filenum = 123, .filename = "filename-goes-here", .force_eof = FALSE}; context_t* pctx = &ctx; rval_evaluator_t* pnr = rval_evaluator_alloc_from_NR(); @@ -65,7 +65,7 @@ static char * test_caps() { static char * test_strings() { printf("\n"); printf("-- TEST_RVAL_EVALUATORS test_strings ENTER\n"); - context_t ctx = {.nr = 888, .fnr = 999, .filenum = 123, .filename = "filename-goes-here"}; + context_t ctx = {.nr = 888, .fnr = 999, .filenum = 123, .filename = "filename-goes-here", .force_eof = FALSE}; context_t* pctx = &ctx; rval_evaluator_t* ps = rval_evaluator_alloc_from_field_name("s", TYPE_INFER_STRING_FLOAT_INT); @@ -131,7 +131,7 @@ static char * test_strings() { static char * test_numbers() { printf("\n"); printf("-- TEST_RVAL_EVALUATORS test_numbers ENTER\n"); - context_t ctx = {.nr = 888, .fnr = 999, .filenum = 123, .filename = "filename-goes-here"}; + context_t ctx = {.nr = 888, .fnr = 999, .filenum = 123, .filename = "filename-goes-here", .force_eof = FALSE}; context_t* pctx = &ctx; rval_evaluator_t* p2 = rval_evaluator_alloc_from_strnum_literal("2.0", TYPE_INFER_STRING_FLOAT_INT); @@ -238,7 +238,7 @@ static char * test_numbers() { static char * test_logical_and() { printf("\n"); printf("-- TEST_RVAL_EVALUATORS test4 ENTER\n"); - context_t ctx = {.nr = 888, .fnr = 999, .filenum = 123, .filename = "filename-goes-here"}; + context_t ctx = {.nr = 888, .fnr = 999, .filenum = 123, .filename = "filename-goes-here", .force_eof = FALSE}; context_t* pctx = &ctx; lrec_t* prec = NULL; @@ -303,7 +303,7 @@ static char * test_logical_and() { static char * test_logical_or() { printf("\n"); printf("-- TEST_RVAL_EVALUATORS test4 ENTER\n"); - context_t ctx = {.nr = 888, .fnr = 999, .filenum = 123, .filename = "filename-goes-here"}; + context_t ctx = {.nr = 888, .fnr = 999, .filenum = 123, .filename = "filename-goes-here", .force_eof = FALSE}; context_t* pctx = &ctx; lrec_t* prec = NULL; @@ -368,7 +368,7 @@ static char * test_logical_or() { static char * test_logical_xor() { printf("\n"); printf("-- TEST_RVAL_EVALUATORS test4 ENTER\n"); - context_t ctx = {.nr = 888, .fnr = 999, .filenum = 123, .filename = "filename-goes-here"}; + context_t ctx = {.nr = 888, .fnr = 999, .filenum = 123, .filename = "filename-goes-here", .force_eof = FALSE}; context_t* pctx = &ctx; lrec_t* prec = NULL;