avro-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dcrea...@apache.org
Subject svn commit: r1221076 - in /avro/trunk: CHANGES.txt lang/c/CMakeLists.txt lang/c/examples/quickstop.c lang/c/src/CMakeLists.txt lang/c/src/avro-c.pc.in lang/c/src/avro/io.h lang/c/src/codec.c lang/c/src/codec.h lang/c/src/datafile.c
Date Tue, 20 Dec 2011 02:27:45 GMT
Author: dcreager
Date: Tue Dec 20 02:27:44 2011
New Revision: 1221076

URL: http://svn.apache.org/viewvc?rev=1221076&view=rev
Log:
AVRO-957. C: Codec support

The C library now supports reading and writing Avro data files that have
compressed data blocks.  We currently support the deflate and lzma
codecs.  The codec API is nice and modular, so it should be easy to add
snappy as well.  Contributed by Michael Cooper and Lucas Martin-King.

Added:
    avro/trunk/lang/c/src/codec.c
    avro/trunk/lang/c/src/codec.h
Modified:
    avro/trunk/CHANGES.txt
    avro/trunk/lang/c/CMakeLists.txt
    avro/trunk/lang/c/examples/quickstop.c
    avro/trunk/lang/c/src/CMakeLists.txt
    avro/trunk/lang/c/src/avro-c.pc.in
    avro/trunk/lang/c/src/avro/io.h
    avro/trunk/lang/c/src/datafile.c

Modified: avro/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/avro/trunk/CHANGES.txt?rev=1221076&r1=1221075&r2=1221076&view=diff
==============================================================================
--- avro/trunk/CHANGES.txt (original)
+++ avro/trunk/CHANGES.txt Tue Dec 20 02:27:44 2011
@@ -33,6 +33,9 @@ Avro 1.6.2 (unreleased)
     AVRO-961. C: avrocat/avropipe can now read from stdin.
     (Michael Cooper via dcreager)
 
+    AVRO-957. C: Codec support in C library.  (Michael Cooper and Lucas
+    Martin-King via dcreager)
+
   BUG FIXES
 
     AVRO-962. Java: Fix Maven plugin to support string type override.

Modified: avro/trunk/lang/c/CMakeLists.txt
URL: http://svn.apache.org/viewvc/avro/trunk/lang/c/CMakeLists.txt?rev=1221076&r1=1221075&r2=1221076&view=diff
==============================================================================
--- avro/trunk/lang/c/CMakeLists.txt (original)
+++ avro/trunk/lang/c/CMakeLists.txt Tue Dec 20 02:27:44 2011
@@ -20,6 +20,10 @@ cmake_minimum_required(VERSION 2.4)
 project(AvroC)
 enable_testing()
 
+# Eliminates warning about linker paths when linking both zlib and
+# liblzma.
+cmake_policy(SET CMP0003 NEW)
+
 #-----------------------------------------------------------------------
 # Retrieve the current version number
 
@@ -69,6 +73,7 @@ string(REGEX REPLACE ".*\\..*\\.[0-9]+(.
 # Source package support
 
 include(CPackConfig.txt)
+include(CheckLibraryExists)
 
 
 if(APPLE)
@@ -87,6 +92,34 @@ endif(CMAKE_COMPILER_IS_GNUCC)
 include_directories(${AvroC_SOURCE_DIR}/src)
 include_directories(${AvroC_SOURCE_DIR}/jansson/src)
 
+
+# Enable codecs
+find_package(ZLIB)
+if (ZLIB_FOUND)
+    set(ZLIB_PKG zlib)
+    add_definitions(-DDEFLATE_CODEC)
+    include_directories(${ZLIB_INCLUDE_DIR})
+    message("Enabled deflate codec")
+else (ZLIB_FOUND)
+    message("Disabled deflate codec. zlib not found.")
+endif (ZLIB_FOUND)
+
+find_package(PkgConfig)
+pkg_check_modules(LZMA liblzma)
+if (LZMA_FOUND)
+    set(LZMA_PKG liblzma)
+    add_definitions(-DLZMA_CODEC)
+    include_directories(${LZMA_INCLUDE_DIRS})
+    link_directories(${LZMA_LIBRARY_DIRS})
+    message("Enabled lzma codec")
+else (LZMA_FOUND)
+    message("Disabled lzma codec. liblzma not found.")
+endif (LZMA_FOUND)
+
+set(CODEC_LIBRARIES ${ZLIB_LIBRARIES} ${LZMA_LIBRARIES})
+set(CODEC_PKG "@ZLIB_PKG@ @LZMA_PKG@")
+
+
 add_subdirectory(src)
 add_subdirectory(examples)
 add_subdirectory(tests)

Modified: avro/trunk/lang/c/examples/quickstop.c
URL: http://svn.apache.org/viewvc/avro/trunk/lang/c/examples/quickstop.c?rev=1221076&r1=1221075&r2=1221076&view=diff
==============================================================================
--- avro/trunk/lang/c/examples/quickstop.c (original)
+++ avro/trunk/lang/c/examples/quickstop.c Tue Dec 20 02:27:44 2011
@@ -64,13 +64,13 @@ add_person(avro_file_writer_t db, const 
 	    || avro_record_set(person, "Last", last_datum)
 	    || avro_record_set(person, "Age", age_datum)
 	    || avro_record_set(person, "Phone", phone_datum)) {
-		fprintf(stderr, "Unable to create Person datum structure");
+		fprintf(stderr, "Unable to create Person datum structure\n");
 		exit(EXIT_FAILURE);
 	}
 
 	if (avro_file_writer_append(db, person)) {
 		fprintf(stderr,
-			"Unable to write Person datum to memory buffer");
+			"Unable to write Person datum to memory buffer\nMessage: %s\n", avro_strerror());
 		exit(EXIT_FAILURE);
 	}
 
@@ -142,9 +142,10 @@ int main(void)
 	/* Delete the database if it exists */
 	unlink(dbname);
 	/* Create a new database */
-	rval = avro_file_writer_create(dbname, person_schema, &db);
+	rval = avro_file_writer_create_with_codec(dbname, person_schema, &db, "deflate", 0);
 	if (rval) {
 		fprintf(stderr, "There was an error creating %s\n", dbname);
+		fprintf(stderr, " error message: %s\n", avro_strerror());
 		exit(EXIT_FAILURE);
 	}
 
@@ -159,6 +160,11 @@ int main(void)
 		add_person(db, "Bob", "Silent", "(555) 123-6422", 29);
 		add_person(db, "Jay", "???", number, 26);
 	}
+
+	/* Close the block and open a new one */
+	avro_file_writer_flush(db);
+	add_person(db, "Super", "Man", "123456", 31);
+
 	avro_file_writer_close(db);
 
 	fprintf(stdout, "\nNow let's read all the records back out\n");
@@ -170,7 +176,7 @@ int main(void)
 	}
 	for (i = 0; i < id; i++) {
 		if (print_person(dbreader, NULL)) {
-			fprintf(stderr, "Error printing person\n");
+			fprintf(stderr, "Error printing person\nMessage: %s\n", avro_strerror());
 			exit(EXIT_FAILURE);
 		}
 	}

Modified: avro/trunk/lang/c/src/CMakeLists.txt
URL: http://svn.apache.org/viewvc/avro/trunk/lang/c/src/CMakeLists.txt?rev=1221076&r1=1221075&r2=1221076&view=diff
==============================================================================
--- avro/trunk/lang/c/src/CMakeLists.txt (original)
+++ avro/trunk/lang/c/src/CMakeLists.txt Tue Dec 20 02:27:44 2011
@@ -35,6 +35,8 @@ set(AVRO_SRC
     avro/value.h
     avro_generic_internal.h
     avro_private.h
+    codec.c
+    codec.h
     consumer.c
     consume-binary.c
     datafile.c
@@ -100,8 +102,10 @@ source_group(Jansson FILES ${JANSSON_SRC
 string(REPLACE ":" "." LIBAVRO_DOT_VERSION ${LIBAVRO_VERSION})
 
 add_library(avro-static STATIC ${AVRO_SRC} ${JANSSON_SRC})
+target_link_libraries(avro-static ${CODEC_LIBRARIES})
 set_target_properties(avro-static PROPERTIES OUTPUT_NAME avro)
 add_library(avro-shared SHARED ${AVRO_SRC} ${JANSSON_SRC})
+target_link_libraries(avro-shared ${CODEC_LIBRARIES})
 set_target_properties(avro-shared PROPERTIES
         OUTPUT_NAME avro
         SOVERSION ${LIBAVRO_DOT_VERSION})
@@ -128,6 +132,7 @@ install(FILES ${CMAKE_CURRENT_BINARY_DIR
         DESTINATION lib/pkgconfig)
 
 add_executable(avrocat avrocat.c)
+
 target_link_libraries(avrocat avro-static)
 install(TARGETS avrocat RUNTIME DESTINATION bin)
 

Modified: avro/trunk/lang/c/src/avro-c.pc.in
URL: http://svn.apache.org/viewvc/avro/trunk/lang/c/src/avro-c.pc.in?rev=1221076&r1=1221075&r2=1221076&view=diff
==============================================================================
--- avro/trunk/lang/c/src/avro-c.pc.in (original)
+++ avro/trunk/lang/c/src/avro-c.pc.in Tue Dec 20 02:27:44 2011
@@ -4,3 +4,4 @@ Version: @VERSION@
 URL: http://avro.apache.org/
 Libs: -L@prefix@/lib -lavro
 Cflags: -I@prefix@/include
+Requires: @CODEC_PKG@

Modified: avro/trunk/lang/c/src/avro/io.h
URL: http://svn.apache.org/viewvc/avro/trunk/lang/c/src/avro/io.h?rev=1221076&r1=1221075&r2=1221076&view=diff
==============================================================================
--- avro/trunk/lang/c/src/avro/io.h (original)
+++ avro/trunk/lang/c/src/avro/io.h Tue Dec 20 02:27:44 2011
@@ -99,6 +99,9 @@ typedef struct avro_file_writer_t_ *avro
 
 int avro_file_writer_create(const char *path, avro_schema_t schema,
 			    avro_file_writer_t * writer);
+int avro_file_writer_create_with_codec(const char *path,
+				avro_schema_t schema, avro_file_writer_t * writer,
+				const char *codec, size_t block_size);
 int avro_file_writer_open(const char *path, avro_file_writer_t * writer);
 int avro_file_reader(const char *path, avro_file_reader_t * reader);
 int avro_file_reader_fp(FILE *fp, const char *path, int should_close,

Added: avro/trunk/lang/c/src/codec.c
URL: http://svn.apache.org/viewvc/avro/trunk/lang/c/src/codec.c?rev=1221076&view=auto
==============================================================================
--- avro/trunk/lang/c/src/codec.c (added)
+++ avro/trunk/lang/c/src/codec.c Tue Dec 20 02:27:44 2011
@@ -0,0 +1,466 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0 
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied.  See the License for the specific language governing
+ * permissions and limitations under the License. 
+ */
+
+#include <string.h>
+#ifdef DEFLATE_CODEC
+#include <zlib.h>
+#endif
+#ifdef LZMA_CODEC
+#include <lzma.h>
+#endif
+#include "avro/errors.h"
+#include "avro/allocation.h"
+#include "codec.h"
+
+#define DEFAULT_BLOCK_SIZE	(16 * 1024)
+
+/* NULL codec */
+
+static int
+codec_null(avro_codec_t codec)
+{
+	codec->name = "null";
+	codec->type = AVRO_CODEC_NULL;
+	codec->block_size = 0;
+	codec->used_size = 0;
+	codec->block_data = NULL;
+	codec->codec_data = NULL;
+
+	return 0;
+}
+
+static int encode_null(avro_codec_t c, void * data, int64_t len)
+{
+	c->block_data = data;
+	c->block_size = len;
+	c->used_size = len;
+
+	return 0;
+}
+
+static int decode_null(avro_codec_t c, void * data, int64_t len)
+{
+	c->block_data = data;
+	c->block_size = len;
+	c->used_size = len;
+
+	return 0;
+}
+
+static int reset_null(avro_codec_t c)
+{
+	c->block_data = NULL;
+	c->block_size = 0;
+	c->used_size = 0;
+	c->codec_data = NULL;
+
+	return 0;
+}
+
+/* Deflate codec */
+
+#ifdef DEFLATE_CODEC
+
+struct codec_data_deflate {
+	z_stream deflate;
+	z_stream inflate;
+};
+#define codec_data_deflate_stream(cd)	&((struct codec_data_deflate *)cd)->deflate
+#define codec_data_inflate_stream(cd)	&((struct codec_data_deflate *)cd)->inflate
+
+
+static int
+codec_deflate(avro_codec_t codec)
+{
+	codec->name = "deflate";
+	codec->type = AVRO_CODEC_DEFLATE;
+	codec->block_size = 0;
+	codec->used_size = 0;
+	codec->block_data = NULL;
+	codec->codec_data = avro_new(struct codec_data_deflate);
+
+	if (!codec->codec_data) {
+		avro_set_error("Cannot allocate memory for zlib");
+		return 1;
+	}
+
+	z_stream *ds = codec_data_deflate_stream(codec->codec_data);
+	z_stream *is = codec_data_inflate_stream(codec->codec_data);
+
+	memset(ds, 0, sizeof(z_stream));
+	memset(is, 0, sizeof(z_stream));
+
+	ds->zalloc = is->zalloc = Z_NULL;
+	ds->zfree  = is->zfree  = Z_NULL;
+	ds->opaque = is->opaque = Z_NULL;
+
+	if (deflateInit2(ds, Z_BEST_COMPRESSION, Z_DEFLATED, -15, 8, Z_DEFAULT_STRATEGY) != Z_OK)
{
+		avro_freet(struct codec_data_deflate, codec->codec_data);
+		avro_set_error("Cannot initialize zlib deflate");
+		return 1;
+	}
+
+	if (inflateInit2(is, -15) != Z_OK) {
+		avro_freet(struct codec_data_deflate, codec->codec_data);
+		avro_set_error("Cannot initialize zlib inflate");
+		return 1;
+	}
+
+	return 0;
+}
+
+static int encode_deflate(avro_codec_t c, void * data, int64_t len)
+{
+	int err;
+	int64_t defl_len = compressBound((uLong)len * 1.2);
+
+	if (!c->block_data) {
+		c->block_data = avro_malloc(defl_len);
+		c->block_size = defl_len;
+	} else if ( c->block_size < defl_len) {
+		c->block_data = avro_realloc(c->block_data, c->block_size, defl_len);
+		c->block_size = defl_len;
+	}
+
+	if (!c->block_data)
+	{
+		avro_set_error("Cannot allocate memory for deflate");
+		return 1;
+	}
+
+	c->used_size = 0;
+
+	z_stream *s = codec_data_deflate_stream(c->codec_data);
+
+	s->next_in = (Bytef*)data;
+	s->avail_in = (uInt)len;
+
+	s->next_out = c->block_data;
+	s->avail_out = (uInt)c->block_size;
+
+	s->total_out = 0;
+
+	err = deflate(s, Z_FINISH);
+	if (err != Z_STREAM_END) {
+		deflateEnd(s);
+		if (err != Z_OK) {
+			avro_set_error("Error compressing block with deflate (%i)", err);
+			return 1;
+		}
+		return 0;
+	}
+
+	// zlib resizes the buffer?
+	c->block_size = s->total_out;
+	c->used_size = s->total_out;
+
+	if (deflateReset(s) != Z_OK) {
+		return 1;
+	}
+
+	return 0;
+}
+
+static int decode_deflate(avro_codec_t c, void * data, int64_t len)
+{
+	int err;
+	z_stream *s = codec_data_inflate_stream(c->codec_data);
+
+	if (!c->block_data) {
+		c->block_data = avro_malloc(DEFAULT_BLOCK_SIZE);
+		c->block_size = DEFAULT_BLOCK_SIZE;
+	}
+
+	if (!c->block_data)
+	{
+		avro_set_error("Cannot allocate memory for deflate");
+		return 1;
+	}
+
+	c->used_size = 0;
+
+	s->next_in = data;
+	s->avail_in = len;
+
+	s->next_out = c->block_data;
+	s->avail_out = c->block_size;
+
+	s->total_out = 0;
+
+	do
+	{
+		err = inflate(s, Z_FINISH);
+
+		// The buffer was not big enough. resize it.
+		if (err == Z_BUF_ERROR)
+		{
+			c->block_data = avro_realloc(c->block_data, c->block_size, c->block_size *
2);
+			s->next_out = c->block_data + s->total_out;
+			s->avail_out += c->block_size;
+			c->block_size = c->block_size * 2;
+		}
+	} while (err == Z_BUF_ERROR);
+
+	if (err != Z_STREAM_END) {
+		inflateEnd(s);
+		if (err != Z_OK) {
+			avro_set_error("Error decompressing block with deflate (%i)", err);
+			return 1;
+		}
+		return 0;
+	}
+
+	c->used_size = s->total_out;
+
+	if (inflateReset(s) != Z_OK) {
+		return 1;
+	}
+
+	return 0;
+}
+
+static int reset_deflate(avro_codec_t c)
+{
+	if (c->block_data) {
+		avro_free(c->block_data, c->block_size);
+	}
+	if (c->codec_data) {
+		deflateEnd(codec_data_deflate_stream(c->codec_data));
+		inflateEnd(codec_data_inflate_stream(c->codec_data));
+		avro_freet(struct codec_data_deflate, c->codec_data);
+	}
+
+	c->block_data = NULL;
+	c->block_size = 0;
+	c->used_size = 0;
+	c->codec_data = NULL;
+
+	return 0;
+}
+
+#endif // DEFLATE_CODEC
+
+/* LZMA codec */
+
+#ifdef LZMA_CODEC
+
+struct codec_data_lzma {
+	lzma_filter filters[2];
+	lzma_options_lzma options;
+};
+#define codec_data_lzma_filters(cd)	((struct codec_data_lzma *)cd)->filters
+#define codec_data_lzma_options(cd)	&((struct codec_data_lzma *)cd)->options
+
+static int
+codec_lzma(avro_codec_t codec)
+{
+	codec->name = "lzma";
+	codec->type = AVRO_CODEC_LZMA;
+	codec->block_size = 0;
+	codec->used_size = 0;
+	codec->block_data = NULL;
+	codec->codec_data = avro_new(struct codec_data_lzma);
+
+	if (!codec->codec_data) {
+		avro_set_error("Cannot allocate memory for lzma");
+		return 1;
+	}
+
+	lzma_options_lzma* opt = codec_data_lzma_options(codec->codec_data);
+	lzma_lzma_preset(opt, LZMA_PRESET_DEFAULT);
+
+	lzma_filter* filters = codec_data_lzma_filters(codec->codec_data);
+	filters[0].id = LZMA_FILTER_LZMA2;
+	filters[0].options = opt;
+	filters[1].id = LZMA_VLI_UNKNOWN;
+	filters[1].options = NULL;
+
+	return 0;
+}
+
+static int encode_lzma(avro_codec_t codec, void * data, int64_t len)
+{
+	lzma_ret ret;
+	size_t written = 0;
+	lzma_filter* filters = codec_data_lzma_filters(codec->codec_data);
+
+	int64_t buff_len = len + lzma_raw_encoder_memusage(filters);
+
+	if (!codec->block_data) {
+		codec->block_data = avro_malloc(buff_len);
+		codec->block_size = buff_len;
+	}
+
+	if (!codec->block_data)
+	{
+		avro_set_error("Cannot allocate memory for lzma encoder");
+		return 1;
+	}
+
+	ret = lzma_raw_buffer_encode(filters, NULL, data, len, codec->block_data, &written,
codec->block_size);
+
+	codec->used_size = written;
+
+	if (ret != LZMA_OK) {
+		avro_set_error("Error in lzma encoder");
+		return 1;
+	}
+
+	return 0;
+}
+
+static int decode_lzma(avro_codec_t codec, void * data, int64_t len)
+{
+	size_t read_pos = 0;
+	size_t write_pos = 0;
+	lzma_ret ret;
+	lzma_filter* filters = codec_data_lzma_filters(codec->codec_data);
+
+	if (!codec->block_data) {
+		codec->block_data = avro_malloc(DEFAULT_BLOCK_SIZE);
+		codec->block_size = DEFAULT_BLOCK_SIZE;
+	}
+
+	if (!codec->block_data) {
+		avro_set_error("Cannot allocate memory for lzma decoder");
+		return 1;
+	}
+
+	do
+	{
+		ret = lzma_raw_buffer_decode(filters, NULL, data,
+			&read_pos, len, codec->block_data, &write_pos,
+			codec->block_size);
+
+		codec->used_size = write_pos;
+
+		// If it ran out of space to decode, give it more!!
+		// It will continue where it left off because of read_pos and write_pos.
+		if (ret == LZMA_BUF_ERROR) {
+			codec->block_data = avro_realloc(codec->block_data, codec->block_size, codec->block_size
* 2);
+			codec->block_size = codec->block_size * 2;
+		}
+
+	} while (ret == LZMA_BUF_ERROR);
+
+	if (ret != LZMA_OK) {
+		avro_set_error("Error in lzma decoder");
+		return 1;
+	}
+
+	return 0;
+}
+
+static int reset_lzma(avro_codec_t c)
+{
+	if (c->block_data) {
+		avro_free(c->block_data, c->block_size);
+	}
+	if (c->codec_data) {
+		avro_freet(struct codec_data_lzma, c->codec_data);
+	}
+
+	c->block_data = NULL;
+	c->block_size = 0;
+	c->used_size = 0;
+	c->codec_data = NULL;
+
+	return 0;
+}
+
+#endif // LZMA_CODEC
+
+/* Common interface */
+
+int avro_codec(avro_codec_t codec, const char *type)
+{
+#ifdef DEFLATE_CODEC
+	if (strcmp("deflate", type) == 0) {
+		return codec_deflate(codec);
+	}
+#endif
+
+#ifdef LZMA_CODEC
+	if (strcmp("lzma", type) == 0) {
+		return codec_lzma(codec);
+	}
+#endif
+
+	if (strcmp("null", type) == 0) {
+		return codec_null(codec);
+	}
+
+	avro_set_error("Unknown codec %s", type);
+	return 1;
+}
+
+int avro_codec_encode(avro_codec_t c, void * data, int64_t len)
+{
+	switch(c->type)
+	{
+	case AVRO_CODEC_NULL:
+		return encode_null(c, data, len);
+#ifdef DEFLATE_CODEC
+	case AVRO_CODEC_DEFLATE:
+		return encode_deflate(c, data, len);
+#endif
+#ifdef LZMA_CODEC
+	case AVRO_CODEC_LZMA:
+		return encode_lzma(c, data, len);
+#endif
+	default:
+		return 1;
+	}
+}
+
+int avro_codec_decode(avro_codec_t c, void * data, int64_t len)
+{
+	switch(c->type)
+	{
+	case AVRO_CODEC_NULL:
+		return decode_null(c, data, len);
+#ifdef DEFLATE_CODEC
+	case AVRO_CODEC_DEFLATE:
+		return decode_deflate(c, data, len);
+#endif
+#ifdef LZMA_CODEC
+	case AVRO_CODEC_LZMA:
+		return decode_lzma(c, data, len);
+#endif
+	default:
+		return 1;
+	}
+}
+
+int avro_codec_reset(avro_codec_t c)
+{
+	switch(c->type)
+	{
+	case AVRO_CODEC_NULL:
+		return reset_null(c);
+#ifdef DEFLATE_CODEC
+	case AVRO_CODEC_DEFLATE:
+		return reset_deflate(c);
+#endif
+#ifdef LZMA_CODEC
+	case AVRO_CODEC_LZMA:
+		return reset_lzma(c);
+#endif
+	default:
+		return 1;
+	}
+}

Added: avro/trunk/lang/c/src/codec.h
URL: http://svn.apache.org/viewvc/avro/trunk/lang/c/src/codec.h?rev=1221076&view=auto
==============================================================================
--- avro/trunk/lang/c/src/codec.h (added)
+++ avro/trunk/lang/c/src/codec.h Tue Dec 20 02:27:44 2011
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0 
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied.  See the License for the specific language governing
+ * permissions and limitations under the License. 
+ */
+
+#ifndef AVRO_CODEC_H
+#define	AVRO_CODEC_H
+
+enum avro_codec_type_t {
+	AVRO_CODEC_NULL,
+	AVRO_CODEC_DEFLATE,
+	AVRO_CODEC_LZMA
+};
+typedef enum avro_codec_type_t avro_codec_type_t;
+
+struct avro_codec_t_ {
+	const char * name;
+	avro_codec_type_t type;
+	int64_t block_size;
+	int64_t used_size;
+	void * block_data;
+	void * codec_data;
+};
+typedef struct avro_codec_t_* avro_codec_t;
+
+int avro_codec(avro_codec_t c, const char *type);
+int avro_codec_reset(avro_codec_t c);
+int avro_codec_encode(avro_codec_t c, void * data, int64_t len);
+int avro_codec_decode(avro_codec_t c, void * data, int64_t len);
+
+#endif

Modified: avro/trunk/lang/c/src/datafile.c
URL: http://svn.apache.org/viewvc/avro/trunk/lang/c/src/datafile.c?rev=1221076&r1=1221075&r2=1221076&view=diff
==============================================================================
--- avro/trunk/lang/c/src/datafile.c (original)
+++ avro/trunk/lang/c/src/datafile.c Tue Dec 20 02:27:44 2011
@@ -21,6 +21,7 @@
 #include "avro/errors.h"
 #include "avro/value.h"
 #include "encoding.h"
+#include "codec.h"
 #include <stdio.h>
 #include <stdlib.h>
 #include <errno.h>
@@ -31,22 +32,29 @@
 struct avro_file_reader_t_ {
 	avro_schema_t writers_schema;
 	avro_reader_t reader;
+	avro_reader_t block_reader;
+	avro_codec_t codec;
 	char sync[16];
 	int64_t blocks_read;
 	int64_t blocks_total;
 	int64_t current_blocklen;
+	char * current_blockdata;
 };
 
 struct avro_file_writer_t_ {
 	avro_schema_t writers_schema;
 	avro_writer_t writer;
+	avro_codec_t codec;
 	char sync[16];
 	int block_count;
 	size_t block_size;
 	avro_writer_t datum_writer;
-	char datum_buffer[16 * 1024];
+	char* datum_buffer;
+	size_t datum_buffer_size;
 };
 
+#define DEFAULT_BLOCK_SIZE 16 * 1024
+
 /* TODO: should we just read /dev/random? */
 static void generate_sync(avro_file_writer_t w)
 {
@@ -82,7 +90,7 @@ static int write_header(avro_file_writer
 	check(rval, enc->write_string(w->writer, "avro.sync"));
 	check(rval, enc->write_bytes(w->writer, w->sync, sizeof(w->sync)));
 	check(rval, enc->write_string(w->writer, "avro.codec"));
-	check(rval, enc->write_bytes(w->writer, "null", 4));
+	check(rval, enc->write_bytes(w->writer, w->codec->name, strlen(w->codec->name)));
 	check(rval, enc->write_string(w->writer, "avro.schema"));
 	schema_writer = avro_writer_memory(schema_buf, sizeof(schema_buf));
 	rval = avro_schema_to_json(w->writers_schema, schema_writer);
@@ -115,7 +123,7 @@ file_writer_init_fp(const char *path, co
 }
 
 static int
-file_writer_create(const char *path, avro_schema_t schema, avro_file_writer_t w)
+file_writer_create(const char *path, avro_schema_t schema, avro_file_writer_t w, size_t block_size)
 {
 	int rval;
 	w->block_count = 0;
@@ -124,11 +132,21 @@ file_writer_create(const char *path, avr
 		check(rval, file_writer_init_fp(path, "w", w));
 	}
 
+	w->datum_buffer_size = block_size;
+	w->datum_buffer = avro_malloc(w->datum_buffer_size);
+
+	if(!w->datum_buffer) {
+		avro_set_error("Could not allocate datum buffer\n");
+		avro_writer_free(w->writer);
+		return ENOMEM;
+	}
+
 	w->datum_writer =
-	    avro_writer_memory(w->datum_buffer, sizeof(w->datum_buffer));
+	    avro_writer_memory(w->datum_buffer, w->datum_buffer_size);
 	if (!w->datum_writer) {
 		avro_set_error("Cannot create datum writer for file %s", path);
 		avro_writer_free(w->writer);
+		avro_free(w->datum_buffer, w->datum_buffer_size);
 		return ENOMEM;
 	}
 
@@ -140,29 +158,52 @@ int
 avro_file_writer_create(const char *path, avro_schema_t schema,
 			avro_file_writer_t * writer)
 {
+	return avro_file_writer_create_with_codec(path, schema, writer, "null", 0);
+}
+
+int avro_file_writer_create_with_codec(const char *path,
+			avro_schema_t schema, avro_file_writer_t * writer,
+			const char *codec, size_t block_size)
+{
 	avro_file_writer_t w;
 	int rval;
 	check_param(EINVAL, path, "path");
 	check_param(EINVAL, is_avro_schema(schema), "schema");
 	check_param(EINVAL, writer, "writer");
+	check_param(EINVAL, codec, "codec");
+
+	if (block_size == 0) {
+		block_size = DEFAULT_BLOCK_SIZE;
+	}
 
 	w = avro_new(struct avro_file_writer_t_);
 	if (!w) {
 		avro_set_error("Cannot allocate new file writer");
 		return ENOMEM;
 	}
-	rval = file_writer_create(path, schema, w);
+	w->codec = avro_new(struct avro_codec_t_);
+	if (!w->codec) {
+		avro_set_error("Cannot allocate new codec");
+		return ENOMEM;
+	}
+	rval = avro_codec(w->codec, codec);
+	if (rval) {
+		avro_freet(struct avro_codec_t_, w->codec);
+		return rval;
+	}
+	rval = file_writer_create(path, schema, w, block_size);
 	if (rval) {
 		avro_freet(struct avro_file_writer_t_, w);
 		return rval;
 	}
 	*writer = w;
+
 	return 0;
 }
 
 static int file_read_header(avro_reader_t reader,
-			    avro_schema_t * writers_schema, char *sync,
-			    int synclen)
+			    avro_schema_t * writers_schema, avro_codec_t codec,
+			    char *sync, int synclen)
 {
 	int rval;
 	avro_schema_t meta_schema;
@@ -170,6 +211,7 @@ static int file_read_header(avro_reader_
 	avro_value_iface_t *meta_iface;
 	avro_value_t meta;
 	char magic[4];
+	avro_value_t codec_val;
 	avro_value_t schema_bytes;
 	const void *p;
 	size_t len;
@@ -196,6 +238,35 @@ static int file_read_header(avro_reader_
 	}
 	avro_schema_decref(meta_schema);
 
+	rval = avro_value_get_by_name(&meta, "avro.codec", &codec_val, NULL);
+	if (rval) {
+		avro_set_error("File header doesn't contain a codec");
+		avro_value_decref(&meta);
+		return rval;
+	} else {
+		const void *buf;
+		size_t size;
+		char codec_name[11];
+
+		avro_type_t type = avro_value_get_type(&codec_val);
+
+		if (type != AVRO_BYTES) {
+			avro_set_error("Value type of codec is unexpected");
+			avro_value_decref(&meta);
+			return EILSEQ;
+		}
+
+		avro_value_get_bytes(&codec_val, &buf, &size);
+		memset(codec_name, 0, sizeof(codec_name));
+		strncpy(codec_name, buf, size < 10 ? size : 10);
+
+		if (avro_codec(codec, codec_name) != 0) {
+			avro_set_error("File header contains an unknown codec");
+			avro_value_decref(&meta);
+			return EILSEQ;
+		}
+	}
+
 	rval = avro_value_get_by_name(&meta, "avro.schema", &schema_bytes, NULL);
 	if (rval) {
 		avro_set_error("File header doesn't contain a schema");
@@ -235,7 +306,7 @@ static int file_writer_open(const char *
 		return ENOMEM;
 	}
 	rval =
-	    file_read_header(reader, &w->writers_schema, w->sync,
+	    file_read_header(reader, &w->writers_schema, w->codec, w->sync,
 			     sizeof(w->sync));
 	avro_reader_free(reader);
 	/* Position to end of file and get ready to write */
@@ -271,11 +342,28 @@ int avro_file_writer_open(const char *pa
 static int file_read_block_count(avro_file_reader_t r)
 {
 	int rval;
+	int64_t len;
 	const avro_encoding_t *enc = &avro_binary_encoding;
 	check_prefix(rval, enc->read_long(r->reader, &r->blocks_total),
 		     "Cannot read file block count: ");
-	check_prefix(rval, enc->read_long(r->reader, &r->current_blocklen),
+	check_prefix(rval, enc->read_long(r->reader, &len),
 		     "Cannot read file block size: ");
+
+	if (r->current_blockdata && len > r->current_blocklen) {
+		r->current_blockdata = avro_realloc(r->current_blockdata, r->current_blocklen,
len);
+		r->current_blocklen = len;
+	} else if (!r->current_blockdata) {
+		r->current_blockdata = avro_malloc(len);
+		r->current_blocklen = len;
+	}
+
+	check_prefix(rval, avro_read(r->reader, r->current_blockdata, len),
+		     "Cannot read file block: ");
+
+	avro_codec_decode(r->codec, r->current_blockdata, len);
+
+	avro_reader_memory_set_source(r->block_reader, r->codec->block_data, r->codec->used_size);
+
 	r->blocks_read = 0;
 	return 0;
 }
@@ -302,15 +390,31 @@ int avro_file_reader_fp(FILE *fp, const 
 		avro_freet(struct avro_file_reader_t_, r);
 		return ENOMEM;
 	}
+	r->block_reader = avro_reader_memory(0, 0);
+	if (!r->block_reader) {
+		avro_set_error("Cannot allocate block reader for file %s", path);
+		avro_freet(struct avro_file_reader_t_, r);
+		return ENOMEM;
+	}
+
+	r->codec = avro_new(struct avro_codec_t_);
+	if (!r->codec) {
+		avro_set_error("Could not allocate codec for file %s", path);
+		avro_freet(struct avro_file_reader_t_, r);
+		return ENOMEM;
+	}
 
-	rval = file_read_header(r->reader, &r->writers_schema, r->sync,
-				sizeof(r->sync));
+	rval = file_read_header(r->reader, &r->writers_schema, r->codec,
+				r->sync, sizeof(r->sync));
 	if (rval) {
 		avro_reader_free(r->reader);
 		avro_freet(struct avro_file_reader_t_, r);
 		return rval;
 	}
 
+	r->current_blockdata = NULL;
+	r->current_blocklen = 0;
+
 	rval = file_read_block_count(r);
 	if (rval) {
 		avro_reader_free(r->reader);
@@ -350,11 +454,14 @@ static int file_write_block(avro_file_wr
 		/* Write the block count */
 		check_prefix(rval, enc->write_long(w->writer, w->block_count),
 			     "Cannot write file block count: ");
+		/* Encode the block */
+		check_prefix(rval, avro_codec_encode(w->codec, w->datum_buffer, w->block_size),
+			     "Cannot encode file block: ");
 		/* Write the block length */
-		check_prefix(rval, enc->write_long(w->writer, w->block_size),
+		check_prefix(rval, enc->write_long(w->writer, w->codec->used_size),
 			     "Cannot write file block size: ");
 		/* Write the block */
-		check_prefix(rval, avro_write(w->writer, w->datum_buffer, w->block_size),
+		check_prefix(rval, avro_write(w->writer, w->codec->block_data, w->codec->used_size),
 			     "Cannot write file block: ");
 		/* Write the sync marker */
 		check_prefix(rval, write_sync(w),
@@ -432,6 +539,9 @@ int avro_file_writer_close(avro_file_wri
 	check(rval, avro_file_writer_flush(w));
 	avro_writer_free(w->datum_writer);
 	avro_writer_free(w->writer);
+	avro_free(w->datum_buffer, w->datum_buffer_size);
+	avro_codec_reset(w->codec);
+	avro_freet(struct avro_codec_t_, w->codec);
 	avro_freet(struct avro_file_writer_t_, w);
 	return 0;
 }
@@ -446,7 +556,7 @@ int avro_file_reader_read(avro_file_read
 	check_param(EINVAL, datum, "datum");
 
 	check(rval,
-	      avro_read_data(r->reader, r->writers_schema, readers_schema,
+	      avro_read_data(r->block_reader, r->writers_schema, readers_schema,
 			     datum));
 	r->blocks_read++;
 
@@ -472,7 +582,7 @@ avro_file_reader_read_value(avro_file_re
 	check_param(EINVAL, r, "reader");
 	check_param(EINVAL, value, "value");
 
-	check(rval, avro_value_read(r->reader, value));
+	check(rval, avro_value_read(r->block_reader, value));
 	r->blocks_read++;
 
 	if (r->blocks_read == r->blocks_total) {
@@ -492,6 +602,12 @@ int avro_file_reader_close(avro_file_rea
 {
 	avro_schema_decref(reader->writers_schema);
 	avro_reader_free(reader->reader);
+	avro_reader_free(reader->block_reader);
+	avro_codec_reset(reader->codec);
+	avro_freet(struct avro_codec_t_, reader->codec);
+	if (reader->current_blockdata) {
+		avro_free(reader->current_blockdata, reader->current_blocklen);
+	}
 	avro_freet(struct avro_file_reader_t_, reader);
 	return 0;
 }



Mime
View raw message