text:
/*
 * cdbparquetstoragewrite.c

 *
 *  Created on: Jul 30, 2013
 *      Author: malili
 */

#include "postgres.h"

#include "catalog/catquery.h"
#include "cdb/cdbparquetstoragewrite.h"
#include "cdb/cdbparquetfooterserializer.h"
#include "lib/stringinfo.h"
#include "utils/cash.h"
#include "utils/geo_decls.h"
#include "utils/date.h"
#include "utils/numeric.h"
#include "utils/xml.h"
#include "utils/inet.h"

#include "snappy-c.h"
#include "zlib.h"

static void addDataPage(
		ParquetColumnChunk columnChunk);

static int encodeCurrentPage(
		ParquetColumnChunk chunk);

static int finalizeCurrentAndNewPage(
		ParquetColumnChunk columnChunk);

static void flushDataPage(
		ParquetColumnChunk chunk,
		int page_number);

static void initGroupType(
		FileField_4C *field,
		char *name,
		RepetitionType repetitionType,
		int hawqType,
		int r,
		int d,
		int depth,
		int numChildren,
		char *parentPathInSchema);

static void initPrimitiveType(
		FileField_4C *field,
		char *name,
		RepetitionType repetitionType,
		int hawqType,
		int typeLen,
		int r,
		int d,
		int depth,
		char *parentPathInSchema);

/* init embedded data types */
static void initPointType(
		FileField_4C *pointField,
		Form_pg_attribute att);

static void initPathType(
		FileField_4C *pathField,
		Form_pg_attribute att);

static void initLsegBoxType(
		FileField_4C *lsegBoxField,
		Form_pg_attribute att);

static void initPolygonType(
		FileField_4C *polygonField,
		Form_pg_attribute att);

static void initCircleType(
		FileField_4C *circleField,
		Form_pg_attribute att);

static void addSingleColumn(
		AppendOnlyEntry *catalog,
		struct ColumnChunkMetadata_4C** columnsMetadata,
		ParquetColumnChunk columns,
		int		*estimateChunkSizes,
		struct	FileField_4C *field,
		int		*colIndex,
		File	parquetFile);

static int appendNullForFields(
		struct FileField_4C *field,
		ParquetColumnChunk columnChunks,
		int *colIndex);

static int appendValueForFields(
		struct FileField_4C *field,
		ParquetColumnChunk columnChunks,
		int *colIndex,
		Datum value);

static int appendParquetColumnNull(
		ParquetColumnChunk columnChunk);

static int appendParquetColumnValue(
		ParquetColumnChunk columnChunk,
		Datum value,
		int r,
		int d);

/*----------------------------------------------------------------
 * append column of geometric types
 * - point, lseg, path, box, polygon, circle
 *----------------------------------------------------------------*/
static int appendParquetColumn_Point(
		ParquetColumnChunk columnChunks,
		int *colIndex,
		Point *point,
		int r,
		int d);

static int appendParquetColumn_Lseg(
		ParquetColumnChunk columnChunks,
		int *colIndex,
		LSEG *lseg,
		int r,
		int d);

static int appendParquetColumn_Path(
		ParquetColumnChunk columnChunks,
		int *colIndex,
		PATH *path,
		int r,
		int d);

static int appendParquetColumn_Box(
		ParquetColumnChunk columnChunks,
		int *colIndex,
		BOX *box,
		int r,
		int d);

static int appendParquetColumn_Polygon(
		ParquetColumnChunk columnChunks,
		int *colIndex,
		POLYGON *polygon,
		int r,
		int d);

static int appendParquetColumn_Circle(
		ParquetColumnChunk columnChunks,
		int *colIndex,
		CIRCLE *circle,
		int r,
		int d);

static char *generateHAWQSchemaStr(
		ParquetFileField pfields,
		int fieldCount);

static void estimateColumnWidths(
		int *columnWidths,
		int ncolumns,
		TupleDesc tableAttrs);

static char *getTypeName(Oid typeOid);

static int encodePlain(
		Datum data,
		ParquetDataPage current_page,
		int hawqTypeId,
		int pageSizeLimit);

static int approximatePageSize(ParquetDataPage page);

static bool ensureBufferCapacity(ParquetDataPage page,
								 int newValueSize,
								 int pageSizeLimit);

#define ENCODE_INVALID_VALUE	-1
#define ENCODE_OUTOF_PAGE		-2

/**
 * generate hawq schema in to string. for example:
 *
 *	message person {
 * 		required varchar name;
 * 		required int2 age;
 * 		optional group home_addr (address) {
 * 			required varchar street;
 * 			required varchar city;
 * 			required varchar state;
 * 			required int4 zip;
 * 		}
 * 		required varchar[] tags;
 * 	}

 What about Array?
 */
static char *
generateHAWQSchemaStr(ParquetFileField pfields,
					  int fieldCount)
{
	StringInfoData schemaBuf;
	initStringInfo(&schemaBuf);
	appendStringInfo(&schemaBuf, "message hawqschema {");

	for (ParquetFileField field = pfields; field < pfields + fieldCount; field++)
	{
		/* TODO add ARRAY and UDF type support */
		char *typeName = getTypeName(field->hawqTypeId);
		appendStringInfo(&schemaBuf, "%s %s %s;",
						 (field->repetitionType == REQUIRED) ? "required" : "optional",
						 typeName,
						 field->name);
		pfree(typeName);
	}

	appendStringInfo(&schemaBuf, "}");
	return schemaBuf.data;
}

int
initparquetMetadata(ParquetMetadata parquetmd,
					TupleDesc tableAttrs,
					File parquetFile)
{
	Form_pg_attribute att;
	ParquetFileField field;

	parquetmd->version = CURRENT_PARQUET_VERSION;
	parquetmd->fieldCount = tableAttrs->natts;
	parquetmd->pfield = palloc0(tableAttrs->natts * sizeof(struct FileField_4C));
	int colCount = 0;
	int schemaTreeNodeCount = 0;
	for (int i = 0; i < tableAttrs->natts; i++)
	{
		att = tableAttrs->attrs[i];
		field = &parquetmd->pfield[i];

		switch (att->atttypid)
		{
		/** basic types */
		case HAWQ_TYPE_BOOL:
		case HAWQ_TYPE_BYTE:
		case HAWQ_TYPE_INT2:
		case HAWQ_TYPE_INT4:
		case HAWQ_TYPE_MONEY:
		case HAWQ_TYPE_INT8:
		case HAWQ_TYPE_FLOAT4:
		case HAWQ_TYPE_FLOAT8:
		case HAWQ_TYPE_NUMERIC:
		/* text related types */
		case HAWQ_TYPE_NAME:
		case HAWQ_TYPE_CHAR:
		case HAWQ_TYPE_BPCHAR:
		case HAWQ_TYPE_VARCHAR:
		case HAWQ_TYPE_TEXT:
		case HAWQ_TYPE_XML:
		/* time related types */
		case HAWQ_TYPE_DATE:
		case HAWQ_TYPE_TIME:
		case HAWQ_TYPE_TIMETZ:
		case HAWQ_TYPE_TIMESTAMP:
		case HAWQ_TYPE_TIMESTAMPTZ:
		case HAWQ_TYPE_INTERVAL:
		/* other types */
		case HAWQ_TYPE_MACADDR:
		case HAWQ_TYPE_INET:
		case HAWQ_TYPE_CIDR:
		case HAWQ_TYPE_BIT:
		case HAWQ_TYPE_VARBIT:
			initPrimitiveType(field,
							  NameStr(att->attname), /* field name*/
							  att->attnotnull ? REQUIRED : OPTIONAL, /* repetition */
							  att->atttypid,	/* HAWQ type */
							  att->attlen,		/* type len */
							  0,						/* r */
							  att->attnotnull ? 0 : 1,	/* d */
							  1,						/* depth */
							  NULL);	/* parent field path */
			colCount += 1;
			schemaTreeNodeCount += 1;
			break;
		/** embedded types */
		case HAWQ_TYPE_POINT:
			initPointType(field, att);
			colCount += 2;
			schemaTreeNodeCount += 3;
			break;
		case HAWQ_TYPE_PATH:
			initPathType(field, att);
			colCount += 3;
			schemaTreeNodeCount += 5;
			break;
		case HAWQ_TYPE_LSEG:
		case HAWQ_TYPE_BOX:
			initLsegBoxType(field, att);
			colCount += 4;
			schemaTreeNodeCount += 5;
			break;
		case HAWQ_TYPE_POLYGON:
			initPolygonType(field, att);
			colCount += 6;
			schemaTreeNodeCount += 9;
			break;
		case HAWQ_TYPE_CIRCLE:
			initCircleType(field, att);
			colCount += 3;
			schemaTreeNodeCount += 4;
			break;
		default:
			ereport(ERROR,
					(errcode(ERRCODE_GP_FEATURE_NOT_SUPPORTED),
					 errmsg("unsupport type '%s'", NameStr(att->attname))));
			break;
		}
	}
	parquetmd->colCount = colCount;
	parquetmd->schemaTreeNodeCount = schemaTreeNodeCount;

	parquetmd->hawqschemastr = generateHAWQSchemaStr(parquetmd->pfield,
													 parquetmd->fieldCount);
	return 0;
}

/*
 * init primitive type
 */
void initPrimitiveType(struct FileField_4C *field,
					   char *name,
					   enum RepetitionType repetitionType,
					   int hawqType,
					   int typeLen,
					   int r,
					   int d,
					   int depth,
					   char *parentPathInSchema)
{
	int nameLen = strlen(name);
	int pathInSchemaLen;
	/*initialize name*/
	field->name = (char*) palloc0(nameLen + 1);
	strcpy(field->name, name);

	/*initialize pathInSchema, should be parentPathInSchema:Name*/
	if (parentPathInSchema == NULL) {
		pathInSchemaLen = nameLen;
		field->pathInSchema = (char*) palloc0(pathInSchemaLen + 1);
		strcpy(field->pathInSchema, name);
	} else {
		pathInSchemaLen = strlen(parentPathInSchema) + nameLen + 1;
		field->pathInSchema = (char*) palloc0(pathInSchemaLen + 1);
		strcpy(field->pathInSchema, parentPathInSchema);
		strcat(field->pathInSchema, ":");
		strcat(field->pathInSchema, name);
	}

	/*initialize other fields*/
	field->repetitionType = repetitionType;
	field->type = mappingHAWQType(hawqType);
	field->hawqTypeId = hawqType;
	field->typeLength = typeLen;
	field->r = r;
	field->d = d;
	field->depth = depth;
}

/**
 * initialize group type
 */
void initGroupType(struct FileField_4C *field,
				   char *name,
				   enum RepetitionType repetitionType,
				   int hawqType,
				   int r,
				   int d,
				   int depth,
				   int numChildren,
				   char *parentPathInSchema)
{
	int nameLen = strlen(name);
	int pathInSchemaLen;
	/*initialize name*/
	field->name = (char *) palloc0 (nameLen + 1);
	strcpy(field->name, name);

	/*initialize pathInSchema, should be parentPathInSchema:Name*/
	if (parentPathInSchema == NULL) {
		pathInSchemaLen = nameLen;
		field->pathInSchema = (char*) palloc0(pathInSchemaLen + 1);
		strcpy(field->pathInSchema, name);
	} else {
		pathInSchemaLen = strlen(parentPathInSchema) + nameLen + 1;
		field->pathInSchema = (char*) palloc0(pathInSchemaLen + 1);
		strcpy(field->pathInSchema, parentPathInSchema);
		strcat(field->pathInSchema, ":");
		strcat(field->pathInSchema, name);
	}

	/*initialize other fields*/
	field->repetitionType = repetitionType;
	field->hawqTypeId = hawqType;
	field->r = r;
	field->d = d;
	field->depth = depth;
	field->num_children = numChildren;
	field->children =
			(struct FileField_4C*) palloc0 (sizeof(struct FileField_4C) * field->num_children);
}

/**
 * initPointType
 * point {required double x; required double y}
 */
void initPointType(struct FileField_4C *pointField, Form_pg_attribute att) {
	/*point itself*/
	enum RepetitionType repetitonType = att->attnotnull ? REQUIRED : OPTIONAL;
	int r = 0;
	int d = att->attnotnull ? 0 : 1;
	int depth = 1;
	int numChildren = 2;
	char *parentPathInSchema = NULL;

	initGroupType(pointField, NameStr(att->attname), repetitonType, att->atttypid,
				  r, d, depth,
				  numChildren, parentPathInSchema);

	/*point:x*/
	struct FileField_4C *child_0 = &(pointField->children[0]);
	initPrimitiveType(child_0, "x", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  pointField->r, pointField->d, pointField->depth + 1,
					  pointField->pathInSchema);

	/*point:y*/
	struct FileField_4C *child_1 = &(pointField->children[1]);
	initPrimitiveType(child_1, "y", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  pointField->r, pointField->d, pointField->depth + 1,
					  pointField->pathInSchema);
}

/**
 * initPathType
 * path:	group {	required boolean is_open;
 * 					repeated group points {required double x; required double y;}}
 */
void initPathType(struct FileField_4C* pathField, Form_pg_attribute att) {
	/*path itself*/
	enum RepetitionType repetitionType = att->attnotnull ? REQUIRED : OPTIONAL;
	int r = 0;
	int d = att->attnotnull ? 0 : 1;
	int depth = 1;
	int numChildren = 2;
	char *parentPathInSchema = NULL;
	initGroupType(pathField, NameStr(att->attname), repetitionType, att->atttypid,
				  r, d, depth,
				  numChildren, parentPathInSchema);

	/* path:is_open */
	struct FileField_4C *child_0 = &(pathField->children[0]);
	initPrimitiveType(child_0, "is_open", REQUIRED, HAWQ_TYPE_BOOL, /*FIXME is bool typeLen 1?*/1,
					  pathField->r, pathField->d, pathField->depth + 1,
					  pathField->pathInSchema);

	/* path:points */
	struct FileField_4C *child_1 = &(pathField->children[1]);
	initGroupType(child_1, "points", REPEATED, HAWQ_TYPE_POINT,
				  pathField->r + 1, pathField->d + 1, pathField->depth + 1,
				  2/*numChildren*/, pathField->pathInSchema);

	/* path:points:x */
	struct FileField_4C * child_1_0 = &(child_1->children[0]);
	initPrimitiveType(child_1_0, "x", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  child_1->r, child_1->d, child_1->depth + 1,
					  child_1->pathInSchema);

	/* path:points:y */
	struct FileField_4C * child_1_1 = &(child_1->children[1]);
	initPrimitiveType(child_1_1, "y", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  child_1->r, child_1->d, child_1->depth + 1,
					  child_1->pathInSchema);
}

/**
 * initLsegBoxType
 * lseg(601)			group {required double x1; required double y1; required double x2; required double y2;}
 *
 * box(603)				group {required double x1; required double y1; required double x2; required double y2;}
 *
 */
void initLsegBoxType(struct FileField_4C *lsegBoxField, Form_pg_attribute att) {
	/*lseg/box itself*/
	enum RepetitionType repetitionType = att->attnotnull ? REQUIRED : OPTIONAL;
	int r = 0;
	int d = att->attnotnull ? 0 : 1;
	int depth = 1;
	int numChildren = 4;
	char *parentPathInSchema = NULL;
	initGroupType(lsegBoxField, NameStr(att->attname), repetitionType, att->atttypid,
				  r, d, depth,
				  numChildren, parentPathInSchema);

	struct FileField_4C *child_0 = &(lsegBoxField->children[0]);
	initPrimitiveType(child_0, "x1", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  lsegBoxField->r, lsegBoxField->d, lsegBoxField->depth + 1,
					  lsegBoxField->pathInSchema);

	struct FileField_4C *child_1 = &(lsegBoxField->children[1]);
	initPrimitiveType(child_1, "y1", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  lsegBoxField->r, lsegBoxField->d, lsegBoxField->depth + 1,
					  lsegBoxField->pathInSchema);

	struct FileField_4C *child_2 = &(lsegBoxField->children[2]);
	initPrimitiveType(child_2, "x2", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  lsegBoxField->r, lsegBoxField->d, lsegBoxField->depth + 1,
					  lsegBoxField->pathInSchema);

	struct FileField_4C *child_3 = &(lsegBoxField->children[3]);
	initPrimitiveType(child_3, "y2", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  lsegBoxField->r, lsegBoxField->d, lsegBoxField->depth + 1,
					  lsegBoxField->pathInSchema);
}

/**
 * init polygon type.
 * group {
 *     required group boundbox {
 *         required double x1;
 *         required double y1;
 *         required double x2;
 *         required double y2;
 *     },
 *     repeated group points {
 *         required double x;
 *         required double y;
 *     }
 * }
 */
void initPolygonType(FileField_4C *polygonField,Form_pg_attribute att)
{
	FileField_4C *box, *points, *f;
	RepetitionType repetitionType = att->attnotnull ? REQUIRED : OPTIONAL;
	int r = 0;
	int d = att->attnotnull ? 0 : 1;
	int depth = 1;
	int numChildren = 2;
	char *parentPathInSchema = NULL;

	/* polygon is a group */
	initGroupType(polygonField, NameStr(att->attname), repetitionType, att->atttypid,
				  r, d, depth,
				  numChildren, parentPathInSchema);
	box = polygonField->children;
	points = polygonField->children + 1;

	/* polygon:boundbox */
	initGroupType(box, "boundbox", REQUIRED, HAWQ_TYPE_BOX,
				  polygonField->r, polygonField->d, polygonField->depth + 1,
				  4 /* numchild */, polygonField->pathInSchema);
	/* polygon:points */
	initGroupType(points, "points", REPEATED, HAWQ_TYPE_POINT,
				  polygonField->r + 1, polygonField->d + 1, polygonField->depth + 1,
				  2 /* numchild */, polygonField->pathInSchema);

	/* polygon:boundbox:{x1,y1,x2,y2} */
	f = box->children;
	initPrimitiveType(f, "x1", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  box->r, box->d, box->depth + 1,
					  box->pathInSchema);
	f = box->children + 1;
	initPrimitiveType(f, "y1", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  box->r, box->d, box->depth + 1,
					  box->pathInSchema);
	f = box->children + 2;
	initPrimitiveType(f, "x2", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  box->r, box->d, box->depth + 1,
					  box->pathInSchema);
	f = box->children + 3;
	initPrimitiveType(f, "y2", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  box->r, box->d, box->depth + 1,
					  box->pathInSchema);

	/* polygon:points:{x,y} */
	f = points->children;
	initPrimitiveType(f, "x", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  points->r, points->d, points->depth + 1,
					  points->pathInSchema);
	f = points->children + 1;
	initPrimitiveType(f, "y", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  points->r, points->d, points->depth + 1,
					  points->pathInSchema);
}

/**
 * initCircleType
 *
 * circle(718)			group {required double x; required double y; required double r;}
 *
 */
void initCircleType(struct FileField_4C *circleField, Form_pg_attribute att) {
	/*circle itself*/
	enum RepetitionType repetitionType = att->attnotnull ? REQUIRED : OPTIONAL;
	int r = 0;
	int d = att->attnotnull ? 0 : 1;
	int depth = 1;
	int numChildren = 3;
	char *parentPathInSchema = NULL;
	initGroupType(circleField, NameStr(att->attname), repetitionType, att->atttypid,
				  r, d, depth, numChildren, parentPathInSchema);

	struct FileField_4C *child_0 = &(circleField->children[0]);
	initPrimitiveType(child_0, "x", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  circleField->r, circleField->d, circleField->depth + 1,
					  circleField->pathInSchema);

	struct FileField_4C *child_1 = &(circleField->children[1]);
	initPrimitiveType(child_1, "y", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  circleField->r, circleField->d, circleField->depth + 1,
					  circleField->pathInSchema);

	struct FileField_4C *child_2 = &(circleField->children[2]);
	initPrimitiveType(child_2, "r", REQUIRED, HAWQ_TYPE_FLOAT8, 8,
					  circleField->r, circleField->d, circleField->depth + 1,
					  circleField->pathInSchema);
}

int
mappingHAWQType(int hawqTypeID)
{
	switch (hawqTypeID)
	{
	case HAWQ_TYPE_BOOL:
		return BOOLEAN;

	case HAWQ_TYPE_INT2:
	case HAWQ_TYPE_INT4:
	case HAWQ_TYPE_MONEY:
	case HAWQ_TYPE_DATE:
		return INT32;

	case HAWQ_TYPE_INT8:
	case HAWQ_TYPE_TIME:
	case HAWQ_TYPE_TIMESTAMPTZ:
	case HAWQ_TYPE_TIMESTAMP:
		return INT64;

	case HAWQ_TYPE_FLOAT4:
		return FLOAT;

	case HAWQ_TYPE_FLOAT8:
		return DOUBLE;

	case HAWQ_TYPE_BIT:
	case HAWQ_TYPE_VARBIT:
	case HAWQ_TYPE_BYTE:
	case HAWQ_TYPE_NUMERIC:
	case HAWQ_TYPE_NAME:
	case HAWQ_TYPE_CHAR:
	case HAWQ_TYPE_BPCHAR:
	case HAWQ_TYPE_VARCHAR:
	case HAWQ_TYPE_TEXT:
	case HAWQ_TYPE_XML:
	case HAWQ_TYPE_TIMETZ:
	case HAWQ_TYPE_INTERVAL:
	case HAWQ_TYPE_MACADDR:
	case HAWQ_TYPE_INET:
	case HAWQ_TYPE_CIDR:
		return BINARY;

	default:
		Insist(false);
		return -1;
	}
}

void estimateColumnWidth(int *columnWidths,
					int *colidx,
					Form_pg_attribute att,
					bool expandEmbeddingType){
	switch (att->atttypid)
	{
		/* fixed size type */
		case HAWQ_TYPE_BOOL:
		case HAWQ_TYPE_INT2:
		case HAWQ_TYPE_INT4:
		case HAWQ_TYPE_INT8:
		case HAWQ_TYPE_FLOAT4:
		case HAWQ_TYPE_FLOAT8:
		case HAWQ_TYPE_DATE:
		case HAWQ_TYPE_TIME:
		case HAWQ_TYPE_TIMETZ:
		case HAWQ_TYPE_TIMESTAMP:
		case HAWQ_TYPE_TIMESTAMPTZ:
		case HAWQ_TYPE_INTERVAL:
		case HAWQ_TYPE_NAME:
		case HAWQ_TYPE_MONEY:
		case HAWQ_TYPE_MACADDR:
			Assert(att->attlen > 0);
			columnWidths[(*colidx)++] = att->attlen;
			break;

		/* variable length type */
		case HAWQ_TYPE_CHAR:
		case HAWQ_TYPE_BPCHAR:
			/* for char(n), atttypmod is n + 4 */
			Assert(att->atttypmod > 4);
			columnWidths[(*colidx)++] = att->atttypmod;
			break;
			break;
		case HAWQ_TYPE_VARCHAR:
		case HAWQ_TYPE_TEXT:
		case HAWQ_TYPE_XML:
			if (att->atttypmod > 4)
			{	/* for varchar(n), atttypmod is n + 4 */
				columnWidths[(*colidx)++] = att->atttypmod;
			}
			else
			{	/* for varchar, text, xml */
				columnWidths[(*colidx)++] = 30;
			}
			break;

		case HAWQ_TYPE_BIT:
		case HAWQ_TYPE_VARBIT:
			if (att->atttypmod > 0)
			{	/* for bit(n) and bit varying (n), atttypmod is n,
				 * but we also have 4 bytes binary header */
				columnWidths[(*colidx)++] = 4 + att->atttypmod;
			}
			else
			{
				columnWidths[(*colidx)++] = 20;
			}
			break;

		case HAWQ_TYPE_BYTE:
		case HAWQ_TYPE_NUMERIC:
		case HAWQ_TYPE_INET:
		case HAWQ_TYPE_CIDR:
			columnWidths[(*colidx)++] = 24;
			break;

		/* maps to multiple columns */
		case HAWQ_TYPE_POINT:
			if (expandEmbeddingType) {
				columnWidths[(*colidx)++] = 8; /* x */
				columnWidths[(*colidx)++] = 8; /* y */
			} else {
				columnWidths[(*colidx)++] = 16;
			}
			break;
		case HAWQ_TYPE_PATH:
			if (expandEmbeddingType) {
				columnWidths[(*colidx)++] = 1; /* is_open */
				columnWidths[(*colidx)++] = 24;/* repeated points.x */
				columnWidths[(*colidx)++] = 24;/* repeated points.y */
			} else {
				columnWidths[(*colidx)++] = 49;
			}
			break;
		case HAWQ_TYPE_LSEG:
		case HAWQ_TYPE_BOX:
			if (expandEmbeddingType) {
				columnWidths[(*colidx)++] = 8; /* x1 */
				columnWidths[(*colidx)++] = 8; /* y1 */
				columnWidths[(*colidx)++] = 8; /* x2 */
				columnWidths[(*colidx)++] = 8; /* y2 */
			} else {
				columnWidths[(*colidx)++] = 32;
			}

			break;
		case HAWQ_TYPE_POLYGON:
			if (expandEmbeddingType) {
				columnWidths[(*colidx)++] = 8; /* boundbox.x1 */
				columnWidths[(*colidx)++] = 8; /* boundbox.y1 */
				columnWidths[(*colidx)++] = 8; /* boundbox.x2 */
				columnWidths[(*colidx)++] = 8; /* boundbox.y2 */
				columnWidths[(*colidx)++] = 24;/* repeated points.x */
				columnWidths[(*colidx)++] = 24;/* repeated points.y */
			} else {
				columnWidths[(*colidx)++] = 80;
			}
			break;
		case HAWQ_TYPE_CIRCLE:
			if (expandEmbeddingType) {
				columnWidths[(*colidx)++] = 8; /* x */
				columnWidths[(*colidx)++] = 8; /* y */
				columnWidths[(*colidx)++] = 8; /* r */
			} else {
				columnWidths[(*colidx)++] = 24;
			}
			break;
		default:
			Insist(false);
			break;
	}
}

static void
estimateColumnWidths(int *columnWidths,
					 int ncolumns,
					 TupleDesc tableAttrs)
{
	int colidx = 0;
	for (int i = 0; i < tableAttrs->natts; i++)
	{
		estimateColumnWidth(columnWidths, &colidx, tableAttrs->attrs[i],
				/*expandEmbeddingType*/ true);
	}
	Assert(colidx == ncolumns);

	const int min_column_width = 1;
	const int max_column_width = 100;

	/* make sure columnwidth in reasonable range to avoid some column
	 * got too low or too high weight */
	for (colidx = 0; colidx < ncolumns; colidx++)
	{
		Insist(columnWidths[colidx] >= min_column_width);
		if (columnWidths[colidx] > max_column_width)
		{
			columnWidths[colidx] = max_column_width;
		}
	}
}

ParquetRowGroup
addRowGroup(ParquetMetadata parquetmd,
			TupleDesc tableAttrs,
			AppendOnlyEntry *aoentry,
			File file)
{
	ParquetRowGroup rowgroup = palloc0(sizeof(struct ParquetRowGroup_S));

	/* ParquetRowGroup */
	rowgroup->catalog			= aoentry;
	rowgroup->rowGroupMetadata	= palloc0(sizeof(struct BlockMetadata_4C));
	rowgroup->columnChunkNumber	= parquetmd->colCount;
	rowgroup->columnChunks		= palloc0(parquetmd->colCount * sizeof(struct ParquetColumnChunk_S));
	rowgroup->parquetFile		= file;

	/* RowGroupMetadata */
	RowGroupMetadata rowgroupmd = rowgroup->rowGroupMetadata;
	rowgroupmd->ColChunkCount	= parquetmd->colCount;
	rowgroupmd->columns			= palloc0(parquetmd->colCount * sizeof(struct ColumnChunkMetadata_4C));
	rowgroupmd->rowCount		= 0;
	rowgroupmd->totalByteSize	= 0;

	if (parquetmd->estimateChunkSizes == NULL)
	{
		parquetmd->estimateChunkSizes = (int *)palloc0(parquetmd->colCount * sizeof(int));

		int *columnWidths = (int *)palloc0(parquetmd->colCount * sizeof(int));
		estimateColumnWidths(columnWidths, parquetmd->colCount, tableAttrs);

		double rowWidths = 0.0;
		for (int i = 0; i < parquetmd->colCount; i++)
		{
			rowWidths += columnWidths[i];
		}

		for (int i = 0; i < parquetmd->colCount; i++)
		{
			parquetmd->estimateChunkSizes[i] =
				(int) ((columnWidths[i] * rowgroup->catalog->blocksize * 1.05)/rowWidths);
		}

		pfree(columnWidths);
	}

	/*
	 * init ParquetColumnChunk (including it's metadata) for each column
	 */
	int cIndex = 0;
	for (int i = 0; i < parquetmd->fieldCount; ++i)
	{
		addSingleColumn(rowgroup->catalog,
						&rowgroupmd->columns,	/* for ColumnChunkMetadata */
						rowgroup->columnChunks,	/* for ParquetColumnChunk */
						parquetmd->estimateChunkSizes,
						/*parquet column chunk max size*/
						&parquetmd->pfield[i],
						&cIndex,
						rowgroup->parquetFile);
	}
	Assert(cIndex == parquetmd->colCount);

	/* should increment the row group count of parquet metadata*/
	parquetmd->blockCount++;
	return rowgroup;
}

void
flushRowGroup(ParquetRowGroup rowgroup,
			  ParquetMetadata parquetmd,
			  MirroredAppendOnlyOpen *mirroredOpen,
			  CompactProtocol *footerProtocol,
			  int64 *fileLen,
			  int64 *fileLen_uncompressed)
{
	int bytes_added = 0;
	Assert(rowgroup != NULL);

	/*
	 * Write out column chunks one by one. For each chunk, we do the following:
	 * 1. encode the last page.
	 * 2. write out pages one by one.
	 * 3. write out chunk's metadata after the last page.
	 */
	for (int i = 0; i < rowgroup->columnChunkNumber; i++)
	{
		ParquetColumnChunk chunk	= &rowgroup->columnChunks[i];
		ColumnChunkMetadata chunkmd	= chunk->columnChunkMetadata;

		bytes_added += encodeCurrentPage(chunk);

		/*----------------------------------------------------------------
		 * recompute estimate chunk size based on uncompressed size (excludes header)
		 *----------------------------------------------------------------*/
		parquetmd->estimateChunkSizes[i] = 0;
		for (int pageno = 0; pageno < chunk->pageNumber; ++pageno)
		{
			parquetmd->estimateChunkSizes[i] += chunk->pages[pageno].header->uncompressed_page_size;
		}
		parquetmd->estimateChunkSizes[i] = (int) (parquetmd->estimateChunkSizes[i] * 1.05);

		/*----------------------------------------------------------------
		 * write out pages one by one
		 *----------------------------------------------------------------*/
		chunkmd->firstDataPage = FileNonVirtualTell(rowgroup->parquetFile);
		if (chunkmd->firstDataPage < 0)
		{
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("file tell position error for segment file: %s", strerror(errno)),
					 errdetail("%s", HdfsGetLastError())));
		}
		for (int pageno = 0; pageno < chunk->pageNumber; ++pageno)
		{
			flushDataPage(chunk, pageno);
		}

		/*----------------------------------------------------------------
		 * write out chunk's metadata after the last page
		 *----------------------------------------------------------------*/
		chunkmd->file_offset = FileNonVirtualTell(rowgroup->parquetFile);
		if (chunkmd->file_offset < 0)
		{
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("file tell position error for segment file: %s", strerror(errno)),
					 errdetail("%s", HdfsGetLastError())));
		}

		uint8_t *Thrift_ColumnMetaData_Buf;
		uint32_t Thrift_ColumnMetaData_Len;
		if (writeColumnChunkMetadata(&Thrift_ColumnMetaData_Buf,
									 &Thrift_ColumnMetaData_Len,
									 chunkmd) != 0)
		{
			ereport(ERROR,
					(errcode(ERRCODE_GP_INTERNAL_ERROR),
					 errmsg("failed to serialize column metadata using thrift")));
		}

		bytes_added += Thrift_ColumnMetaData_Len;

		if (FileWrite(rowgroup->parquetFile,
					  (char *) Thrift_ColumnMetaData_Buf,
					  Thrift_ColumnMetaData_Len) != Thrift_ColumnMetaData_Len)
		{
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("file write error when writing out column metadata: %s", strerror(errno)),
					 errdetail("%s", HdfsGetLastError())));
		}

		/* Add chunk compressedsize and uncompressedsize to parquet fileLen and fileLen_uncompressed*/
		(*fileLen) += (chunkmd->totalSize + Thrift_ColumnMetaData_Len);
		(*fileLen_uncompressed) += (chunkmd->totalUncompressedSize + Thrift_ColumnMetaData_Len);
	}

	int fileSync = 0;
	MirroredAppendOnly_Flush(mirroredOpen, &fileSync);
	if(fileSync < 0){
		ereport(ERROR,
				(errcode_for_file_access(),
						errmsg("file sync error: %s", strerror(fileSync)),
						errdetail("%s", HdfsGetLastError())));
	}

	rowgroup->rowGroupMetadata->totalByteSize += bytes_added;
	parquetmd->num_rows += rowgroup->rowGroupMetadata->rowCount;

	writeRowGroupInfo(rowgroup->rowGroupMetadata, footerProtocol);
	freeRowGroup(rowgroup);
}

void
freeRowGroup(ParquetRowGroup rowgroup)
{
	for (int i = 0; i < rowgroup->columnChunkNumber; i++)
	{
		pfree(rowgroup->columnChunks[i].pages);

		/* chunk metadata should be kept util parquet_insert_finish */
		rowgroup->columnChunks[i].columnChunkMetadata = NULL;
	}
	pfree(rowgroup->columnChunks);

	freeRowGroupInfo(rowgroup->rowGroupMetadata);
	pfree(rowgroup);
}

/**
 * add a column information to row group
 * catalog:				pg_appendonly entry
 * columnsMetadata:		columnMetadata needed to be added
 * columns:				column chunks of row group needed to be initialized
 * estimateChunkSizes:	array of estimated sizes for each columnchunk
 * field:				the column description in parquet file metadata schema part
 * colIndex:			the index of the column in columnsMetadata and columns
 * parquetFile:			file to insert into
 */
void
addSingleColumn(AppendOnlyEntry *catalog,
				struct ColumnChunkMetadata_4C** columnsMetadata,
				ParquetColumnChunk columns,
				int		*estimateChunkSizes,
				struct	FileField_4C *field,
				int		*colIndex,
				File	parquetFile)
{
	if (field->num_children > 0)
	{	/* for embedded types, should expand it, recursive call the function itself*/
		for (int i = 0; i < field->num_children; i++) {
			addSingleColumn(catalog, columnsMetadata, columns, estimateChunkSizes,
							&(field->children[i]), colIndex, parquetFile);
		}
	}
	else
	{	/* for single column, directly add the column*/
		struct ColumnChunkMetadata_4C* chunkmd =
				&((*columnsMetadata)[*colIndex]);

		/*----------------------------------------------------------------
		 * initialize ColumnChunkMetadata
		 *----------------------------------------------------------------*/
		chunkmd->EncodingCount			= 3;
		chunkmd->pEncodings 			= palloc0(chunkmd->EncodingCount * sizeof(enum Encoding));
		chunkmd->pEncodings[0] 			= RLE; /*set definition level encoding as RLE*/
		chunkmd->pEncodings[1] 			= RLE; /*set repetition level encoding as RLE*/
		chunkmd->pEncodings[2] 			= PLAIN; /*set data encoding as PLAIN*/
		chunkmd->file_offset 			= 0;
		chunkmd->firstDataPage 			= 0;
		chunkmd->totalSize 				= 0;
		chunkmd->totalUncompressedSize 	= 0;
		chunkmd->valueCount 			= 0;

		if (catalog->compresstype == NULL)
		{
			chunkmd->codec = UNCOMPRESSED;
		}
		else
		{
			if (0 == strcmp(catalog->compresstype, "snappy"))
			{
				chunkmd->codec = SNAPPY;
			}
			else if (0 == strcmp(catalog->compresstype, "gzip"))
			{
				chunkmd->codec = GZIP;
			}
#ifdef NOT_USED
			else if (0 == strcmp(catalog->compresstype, "lzo"))
			{
				chunkmd->codec = LZO;
			}
#endif
			else
			{
				Assert(0 == strcmp(catalog->compresstype, "none"));
				chunkmd->codec = UNCOMPRESSED;
			}
		}

		chunkmd->type 			= field->type;
		chunkmd->hawqTypeId 	= field->hawqTypeId;
		chunkmd->colName 		= (char *)palloc0(strlen(field->name) + 1);
		strcpy(chunkmd->colName, field->name);
		chunkmd->pathInSchema 	= (char *)palloc0(strlen(field->pathInSchema) + 1);
		strcpy(chunkmd->pathInSchema, field->pathInSchema);
		chunkmd->r 				= field->r;
		chunkmd->d 				= field->d;
		chunkmd->depth 			= field->depth;

		/*----------------------------------------------------------------
		 * initialize ParquetColumnChunk
		 *----------------------------------------------------------------*/
		ParquetColumnChunk chunk 			= columns + (*colIndex);
		chunk->columnChunkMetadata			= chunkmd;
		chunk->maxPageCount 				= DEFAULT_DATAPAGE_COUNT;
		chunk->pageNumber 					= 0;
		chunk->pages 						= palloc0(chunk->maxPageCount * sizeof(struct ParquetDataPage_S));
		chunk->currentPage					= NULL;
		chunk->estimateChunkSizeRemained 	= estimateChunkSizes[*colIndex];
		chunk->pageSizeLimit 				= catalog->pagesize;
		chunk->compresstype					= catalog->compresstype;
		chunk->compresslevel				= catalog->compresslevel;
		chunk->parquetFile					= parquetFile;

		*colIndex = *colIndex + 1;
	}
}

/*
 * Write out a specified data page (page header + page data)
 */
static void
flushDataPage(ParquetColumnChunk chunk, int page_number)
{
	ParquetDataPage page = &chunk->pages[page_number];
	Assert(page != NULL);
	Assert(page->finalized);
	Assert(page->header_buffer != NULL);

	/*----------------------------------------------------------------
	 * write out thrift page header
	 *----------------------------------------------------------------*/
	if (FileWrite(page->parquetFile,
				  (char *) page->header_buffer,
				  page->header_len) != page->header_len)
	{
		ereport(ERROR,
				(errcode_for_file_access(),
						errmsg("file write error when writing out page header: %s", strerror(errno)),
						errdetail("%s", HdfsGetLastError())));
	}
	pfree(page->header_buffer);

	/*----------------------------------------------------------------
	 * write out page data
	 *----------------------------------------------------------------*/
	if (FileWrite(page->parquetFile,
				  (char *) page->data,
				  page->header->compressed_page_size) != page->header->compressed_page_size)
	{
		ereport(ERROR,
				(errcode_for_file_access(),
						errmsg("file write error when writing out page data: %s", strerror(errno)),
						errdetail("%s", HdfsGetLastError())));
	}

	pfree(page->header);
	if (page->data != NULL)
	{
		pfree(page->data);
	}
}

/*
 * Append value to the corresponding data page.
 *
 * Upon successful completion, the number of bytes which were added is returned.
 *
 * Otherwise ENCODE_INVALID_VALUE is returned if encoded length of `data`
 * exceeds `pageSizeLimit`.
 *
 * ENCODE_OUTOF_PAGE is returned if `data` is of valid size but appending
 * the data will make the page exceeds `pageSizeLimit`.
 *
 */
static int
encodePlain(Datum data,
			ParquetDataPage current_page,
			int hawqTypeId,
			int pageSizeLimit)
{
	int len = 0; /* actual number of bytes added to buffer */
	uint8_t* dst_ptr = NULL;

	switch (hawqTypeId)
	{

	case HAWQ_TYPE_BOOL:
	{
		if (approximatePageSize(current_page) >= pageSizeLimit)
		{
			return ENCODE_OUTOF_PAGE;
		}
		return BitPack_WriteInt(current_page->bool_values, DatumGetBool(data) ? 1 : 0);
	}

	/*----------------------------------------------------------------
	 * Type mapped to 4-bytes INT32/FLOAT in Parquet
	 *----------------------------------------------------------------*/
	case HAWQ_TYPE_INT2:
	{
		len = 4;
		if (!ensureBufferCapacity(current_page, len, pageSizeLimit))
		{
			return ENCODE_OUTOF_PAGE;
		}
		int32 val = (int32) DatumGetInt16(data);

		dst_ptr = current_page->values_buffer +
				  current_page->header->uncompressed_page_size;

		memcpy(dst_ptr, &val, len);
		return len;
	}
	case HAWQ_TYPE_INT4:
	{
		len = 4;
		if (!ensureBufferCapacity(current_page, len, pageSizeLimit))
		{
			return ENCODE_OUTOF_PAGE;
		}
		int32 val = DatumGetInt32(data);

		dst_ptr = current_page->values_buffer +
				  current_page->header->uncompressed_page_size;

		memcpy(dst_ptr, &val, len);
		return len;
	}
	case HAWQ_TYPE_DATE:
	{
		len = 4;
		if (!ensureBufferCapacity(current_page, len, pageSizeLimit))
		{
			return ENCODE_OUTOF_PAGE;
		}
		DateADT val = DatumGetDateADT(data);

		dst_ptr = current_page->values_buffer +
				  current_page->header->uncompressed_page_size;

		memcpy(dst_ptr, &val, len);
		return len;
	}
	case HAWQ_TYPE_FLOAT4:
	{
		len = 4;
		if (!ensureBufferCapacity(current_page, len, pageSizeLimit))
		{
			return ENCODE_OUTOF_PAGE;
		}
		float4 val = DatumGetFloat4(data);

		dst_ptr = current_page->values_buffer +
				  current_page->header->uncompressed_page_size;

		memcpy(dst_ptr, &val, len);
		return len;
	}
	case HAWQ_TYPE_MONEY:
	{
		/*
		 * Although money is represented as int32 internally,
		 * it's passed by reference.
		 */
		len = 4;
		if (!ensureBufferCapacity(current_page, len, pageSizeLimit))
		{
			return ENCODE_OUTOF_PAGE;
		}
		Cash *cash_p = (Cash *) DatumGetPointer(data);

		dst_ptr = current_page->values_buffer +
				  current_page->header->uncompressed_page_size;

		memcpy(dst_ptr, cash_p, len);
		return len;
	}

	/*----------------------------------------------------------------
	 * Type mapped to 8-bytes INT64/DOUBLE in Parquet
	 *----------------------------------------------------------------*/
	case HAWQ_TYPE_INT8:
	{
		len = 8;
		if (!ensureBufferCapacity(current_page, len, pageSizeLimit))
		{
			return ENCODE_OUTOF_PAGE;
		}
		int64 val = DatumGetInt64(data);

		dst_ptr = current_page->values_buffer +
				  current_page->header->uncompressed_page_size;

		memcpy(dst_ptr, &val, len);
		return len;
	}
	case HAWQ_TYPE_TIME:
	{
		len = 8;
		if (!ensureBufferCapacity(current_page, len, pageSizeLimit))
		{
			return ENCODE_OUTOF_PAGE;
		}
		TimeADT val = DatumGetTimeADT(data);

		dst_ptr = current_page->values_buffer +
				  current_page->header->uncompressed_page_size;

		memcpy(dst_ptr, &val, len);
		return len;
	}
	case HAWQ_TYPE_TIMESTAMP:
	{
		len = 8;
		if (!ensureBufferCapacity(current_page, len, pageSizeLimit))
		{
			return ENCODE_OUTOF_PAGE;
		}
		Timestamp val = DatumGetTimestamp(data);

		dst_ptr = current_page->values_buffer +
				  current_page->header->uncompressed_page_size;

		memcpy(dst_ptr, &val, len);
		return len;
	}
	case HAWQ_TYPE_TIMESTAMPTZ:
	{
		len = 8;
		if (!ensureBufferCapacity(current_page, len, pageSizeLimit))
		{
			return ENCODE_OUTOF_PAGE;
		}
		TimestampTz val = DatumGetTimestampTz(data);

		dst_ptr = current_page->values_buffer +
				  current_page->header->uncompressed_page_size;

		memcpy(dst_ptr, &val, len);
		return len;
	}
	case HAWQ_TYPE_FLOAT8:
	{
		len = 8;
		if (!ensureBufferCapacity(current_page, len, pageSizeLimit))
		{
			return ENCODE_OUTOF_PAGE;
		}
		float8 val = DatumGetFloat8(data);

		dst_ptr = current_page->values_buffer +
				  current_page->header->uncompressed_page_size;

		memcpy(dst_ptr, &val, len);
		return len;
	}

	/*----------------------------------------------------------------
	 * fixed length type, mapped to BINARY in Parquet
	 *----------------------------------------------------------------*/
	case HAWQ_TYPE_NAME:
	{
		int data_size = NAMEDATALEN;
		NameData *name = DatumGetName(data);
		len = 4 + data_size;
		if (!ensureBufferCapacity(current_page, len, pageSizeLimit))
		{
			return ENCODE_OUTOF_PAGE;
		}

		dst_ptr = current_page->values_buffer +
				  current_page->header->uncompressed_page_size;

		memcpy(dst_ptr, &(/*htole32(*/data_size/*)*/), 4);
		dst_ptr += 4;
		memcpy(dst_ptr, NameStr(*name), data_size);
		return len;
	}
	case HAWQ_TYPE_TIMETZ:
	{
		/*
		 * timetz (12 bytes) is stored in parquet's BINARY type,
		 * that is <4-bytes-header> + <12-bytes-content>
		 */
		int data_size = sizeof(TimeTzADT);
		TimeTzADT *timetz = DatumGetTimeTzADTP(data);
		len = 4 + data_size;
		if (!ensureBufferCapacity(current_page, len, pageSizeLimit))
		{
			return ENCODE_OUTOF_PAGE;
		}

		dst_ptr = current_page->values_buffer +
				  current_page->header->uncompressed_page_size;

		memcpy(dst_ptr, &(/*htole32(*/data_size/*)*/), 4);
		dst_ptr += 4;
		memcpy(dst_ptr, timetz, sizeof(TimeTzADT));
		return len;
	}
	case HAWQ_TYPE_INTERVAL:
	{
		int data_size = sizeof(Interval);
		Interval *interval = DatumGetIntervalP(data);
		len = 4 + data_size;
		if (!ensureBufferCapacity(current_page, len, pageSizeLimit))
		{
			return ENCODE_OUTOF_PAGE;
		}

		dst_ptr = current_page->values_buffer +
				  current_page->header->uncompressed_page_size;

		memcpy(dst_ptr, &(/*htole32(*/data_size/*)*/), 4);
		dst_ptr += 4;
		memcpy(dst_ptr, interval, sizeof(Interval));
		return len;
	}
	case HAWQ_TYPE_MACADDR:
	{
		int data_size = 6;
		macaddr *mac = DatumGetMacaddrP(data);
		len = 4 + data_size;
		if (!ensureBufferCapacity(current_page, len, pageSizeLimit))
		{
			return ENCODE_OUTOF_PAGE;
		}

		dst_ptr = current_page->values_buffer +
				  current_page->header->uncompressed_page_size;

		memcpy(dst_ptr, &(/*htole32(*/data_size/*)*/), 4);
		dst_ptr += 4;
		/* TODO can we just memcpy the structure? */
		memcpy(dst_ptr, &mac->a, sizeof(char));
		dst_ptr += sizeof(char);
		memcpy(dst_ptr, &mac->b, sizeof(char));
		dst_ptr += sizeof(char);
		memcpy(dst_ptr, &mac->c, sizeof(char));
		dst_ptr += sizeof(char);
		memcpy(dst_ptr, &mac->d, sizeof(char));
		dst_ptr += sizeof(char);
		memcpy(dst_ptr, &mac->e, sizeof(char));
		dst_ptr += sizeof(char);
		memcpy(dst_ptr, &mac->f, sizeof(char));
		return len;
	}

	/*
	 * variable length type, mapped to BINARY in Parquet
	 * ------------------
	 * The following types are implemented as varlena in HAWQ, they all corresponds
	 * to BINARY in Parquet. However, we have two strategies when storing them, depends
	 * on whether we stores varlena header or not.
	 *
	 * [strategy 1] exclude varlena header:
	 * 				BINARY = [VARSIZE(d) - 4, VARDATA(d)]
	 * 				This strategy works better for text-related type because
	 * 				any parquet client can interprete text binary.
	 *
	 * [strategy 2] include varlena header in actual data:
	 * 				BINARY = [VARSIZE(d), d]
	 * 				This works better for HAWQ specific type like numeric because
	 * 				we can easily deserialize data. (just get the data part of the byte array)
	 */

	/* these types use [strategy 1] */
	case HAWQ_TYPE_BYTE:
	case HAWQ_TYPE_CHAR:
	case HAWQ_TYPE_BPCHAR:
	case HAWQ_TYPE_VARCHAR:
	case HAWQ_TYPE_TEXT:
	case HAWQ_TYPE_XML:
	{
		struct varlena *varlen = (struct varlena *) DatumGetPointer(data);
		Assert(!VARATT_IS_COMPRESSED(varlen) && !VARATT_IS_EXTERNAL(varlen));

		int puredataSize = VARSIZE_ANY_EXHDR(varlen);
		len = puredataSize + 4;

		if (len > pageSizeLimit)
		{
			return ENCODE_INVALID_VALUE;
		}
		if (!ensureBufferCapacity(current_page, len, pageSizeLimit))
		{
			return ENCODE_OUTOF_PAGE;
		}

		dst_ptr = current_page->values_buffer +
				  current_page->header->uncompressed_page_size;

		memcpy(dst_ptr, &(/*htole32(*/puredataSize/*)*/), 4);
		dst_ptr += 4;
		memcpy(dst_ptr, VARDATA_ANY(varlen), puredataSize);
		return len;
	}
	/* these types use [strategy 2] */
	case HAWQ_TYPE_BIT:
	case HAWQ_TYPE_VARBIT:
	case HAWQ_TYPE_NUMERIC:
	case HAWQ_TYPE_INET:
	case HAWQ_TYPE_CIDR:
	{
		struct varlena *varlen = (struct varlena *) DatumGetPointer(data);
		Assert(!VARATT_IS_COMPRESSED(varlen) && !VARATT_IS_EXTERNAL(varlen));

		int dataSize = VARSIZE_ANY(varlen);
		len = dataSize + sizeof(int32);

		if (len > pageSizeLimit)
		{
			return ENCODE_INVALID_VALUE;
		}
		if (!ensureBufferCapacity(current_page, len, pageSizeLimit))
		{
			return ENCODE_OUTOF_PAGE;
		}

		dst_ptr = current_page->values_buffer +
				  current_page->header->uncompressed_page_size;

		memcpy(dst_ptr, &(/*htole32(*/dataSize/*)*/), sizeof(int32));
		dst_ptr += sizeof(int32);
		memcpy(dst_ptr, varlen, dataSize);
		return len;
	}
	default:
		Insist(false);
		break;
	}
}

/*
 * Ensure page's values_buffer is large enough to contain the new value.
 *
 * Return false if adding the value will make the page exceeds `pageSizeLimit`.
 * Return true otherwise.
 */
static bool ensureBufferCapacity(ParquetDataPage page,
								 int newValueSize, int pageSizeLimit)
{
	if ((approximatePageSize(page) + newValueSize) > pageSizeLimit)
		return false;

	/* the lower bound size for values_buffer to contain the new value */
	int buffer_lowerbound = page->header->uncompressed_page_size + newValueSize;

	Assert(buffer_lowerbound <= pageSizeLimit);

	/* make sure buffer_lowerbound <= values_buffer_capacity <= pageSizeLimit */
	if (buffer_lowerbound > page->values_buffer_capacity)
	{
		page->values_buffer_capacity = buffer_lowerbound > page->values_buffer_capacity * 2
											? buffer_lowerbound
											: page->values_buffer_capacity * 2;

		if (page->values_buffer_capacity > pageSizeLimit)
			page->values_buffer_capacity = pageSizeLimit;

		page->values_buffer = repalloc(page->values_buffer, page->values_buffer_capacity);
	}
	return true;
}

/**
 * Put the final page data (may be compressed) in page->data, and the
 * final page header data in page->header_buffer.
 *
 * Return added size of uncompressed data in the current row group, which
 * is 'unflushed rle/bitpack data' + 'page header'
 */
int
encodeCurrentPage(ParquetColumnChunk chunk)
{
	int bytes_added;
	ParquetDataPage current_page;
	ParquetPageHeader header;
	ColumnChunkMetadata chunkmd;

	bytes_added = 0;
	current_page = chunk->currentPage;
	header = current_page->header;
	chunkmd = chunk->columnChunkMetadata;

	Assert(current_page != NULL);

	if (current_page->finalized)
		return 0;

	/*----------------------------------------------------------------
	 * Flush RLE/BitPack encoded data. Size of r and d data are
	 * accumulated into page's uncompressed_page_size in this phase.
	 *
	 * r/d = <4-bytes little-endian encoded-data-len> + <encoded-data>
	 *----------------------------------------------------------------*/
	if(current_page->repetition_level != NULL)
	{
		RLEEncoder_Flush(current_page->repetition_level);
		bytes_added += 4 + RLEEncoder_Size(current_page->repetition_level);
	}

	if(current_page->definition_level != NULL)
	{
		RLEEncoder_Flush(current_page->definition_level);
		bytes_added += 4 + RLEEncoder_Size(current_page->definition_level);
	}

	if (chunkmd->hawqTypeId == HAWQ_TYPE_BOOL)
	{
		bytes_added += BitPack_Flush(current_page->bool_values);
	}

	header->uncompressed_page_size += bytes_added;

	/* we must make sure there is no empty page, since some compression algorithm
	 * will fail if input buffer is NULL */
	Assert(header->uncompressed_page_size > 0);

	/*----------------------------------------------------------------
	 * Combine r/d/value bytes into a buffer for compressing.
	 *----------------------------------------------------------------*/
	StringInfoData buf;
	/* we don't want to StringInfo to enlarge its buffer during appendXXX,
	 * however StringInfo has a trailing '\0', so we add 1 here */
	initStringInfoOfSize(&buf, header->uncompressed_page_size + 1);

	if(current_page->repetition_level != NULL)
	{
		int encoded_data_len = RLEEncoder_Size(current_page->repetition_level);
		appendBinaryStringInfo(&buf, &(/*htole32(*/encoded_data_len/*)*/), 4);
		appendBinaryStringInfo(&buf,
							   RLEEncoder_Data(current_page->repetition_level),
							   encoded_data_len);

		pfree(current_page->repetition_level->writer.buffer);
		pfree(current_page->repetition_level->packBuffer);
		pfree(current_page->repetition_level);
	}

	if(current_page->definition_level != NULL)
	{
		int encoded_data_len = RLEEncoder_Size(current_page->definition_level);
		appendBinaryStringInfo(&buf, &(/*htole32(*/encoded_data_len/*)*/), 4);
		appendBinaryStringInfo(&buf,
							   RLEEncoder_Data(current_page->definition_level),
							   encoded_data_len);

		pfree(current_page->definition_level->writer.buffer);
		pfree(current_page->definition_level->packBuffer);
		pfree(current_page->definition_level);
	}

	if (chunkmd->hawqTypeId == HAWQ_TYPE_BOOL)
	{
		appendBinaryStringInfo(&buf,
							   BitPack_Data(current_page->bool_values),
							   BitPack_Size(current_page->bool_values));

		pfree(current_page->bool_values->buffer);
		pfree(current_page->bool_values);
	}
	else
	{
		appendBinaryStringInfo(&buf,
							   current_page->values_buffer,
							   header->uncompressed_page_size - buf.len);

		pfree(current_page->values_buffer);
	}

	/*----------------------------------------------------------------
	 * Compress page data if needed, saved it to current_page->data.
	 *----------------------------------------------------------------*/
	switch (chunkmd->codec)
	{
		case UNCOMPRESSED:
		{
			current_page->data = (uint8_t*) buf.data;
			header->compressed_page_size = header->uncompressed_page_size;
			break;
		}

		case SNAPPY:
		{
			size_t compressedLen = snappy_max_compressed_length(header->uncompressed_page_size);
			current_page->data = (uint8_t *) palloc(compressedLen);

			if (snappy_compress(buf.data, header->uncompressed_page_size,
								(char *)current_page->data, &compressedLen) == SNAPPY_OK)
			{
				pfree(buf.data);
				header->compressed_page_size = compressedLen;
			}
			else
			{
				ereport(ERROR,
					(errcode(ERRCODE_GP_INTERNAL_ERROR),
					 errmsg("snappy compression failed: %s", (char *)current_page->data)));
			}

			break;
		}
		case GZIP:
		{
			int ret;
			/* 15(default windowBits for deflate) + 16(ouput GZIP header/tailer) */
			const int windowbits = 31;

			z_stream stream;
			stream.zalloc	= Z_NULL;
			stream.zfree	= Z_NULL;
			stream.opaque	= Z_NULL;
			stream.avail_in	= header->uncompressed_page_size;
			stream.next_in	= (Bytef *) buf.data;

			ret = deflateInit2(&stream, chunk->compresslevel, Z_DEFLATED,
							   windowbits, MAX_MEM_LEVEL, Z_DEFAULT_STRATEGY);
			if (ret != Z_OK)
			{
				ereport(ERROR,
						(errcode(ERRCODE_GP_INTERNAL_ERROR),
						 errmsg("zlib deflateInit2 failed: %s", stream.msg)));
			}

			size_t compressedLen = header->uncompressed_page_size;
			current_page->data = (uint8_t *) palloc(compressedLen);

			Bytef *out = (Bytef *) current_page->data;
			int outlen = compressedLen;

			/* process until all inputs have been compressed */
			do
			{
				stream.next_out = out;
				stream.avail_out = outlen;

				ret = deflate(&stream, Z_FINISH);
				if (ret == Z_STREAM_END)
					break;
				if (ret == Z_OK)
				{
					/* out buffer is not big enough, extend 4096 byte at a time */
					outlen = 4096;
					current_page->data = repalloc(current_page->data, compressedLen + outlen);
					out = current_page->data + compressedLen;
					compressedLen += outlen;
				}
				else
				{
					deflateEnd(&stream);
					ereport(ERROR,
							(errcode(ERRCODE_GP_INTERNAL_ERROR),
							 errmsg("zlib deflate failed: %s", stream.msg)));
				}

			} while (1);

			compressedLen = stream.total_out;
			deflateEnd(&stream);

			pfree(buf.data);
			header->compressed_page_size = compressedLen;
			break;
		}
#ifdef NOT_USED
		case LZO:
			/* TODO*/
			Insist(false);
			break;
#endif
		default:
			Insist(false);	/* shouldn't get here */
			break;
	}

	/*----------------------------------------------------------------
	 * All fields of page header are filled, convert to binary page
	 * header in thrift.
	 *----------------------------------------------------------------*/
	uint8_t* header_buffer = NULL;
	if (writePageMetadata(&header_buffer, (uint32_t *) &current_page->header_len,
					  current_page->header) < 0)
	{
		ereport(ERROR,
				(errcode(ERRCODE_GP_INTERNAL_ERROR),
				 errmsg("failed to serialize page metadata using thrift for column: %s", chunkmd->colName)));
	}

	current_page->header_buffer = (uint8_t *) palloc0(current_page->header_len);
	memcpy(current_page->header_buffer, header_buffer, current_page->header_len);

	chunkmd->totalUncompressedSize	+= current_page->header_len + header->uncompressed_page_size;
	chunkmd->totalSize				+= current_page->header_len + header->compressed_page_size;

	current_page->finalized = true;

	return bytes_added;
}

static void
addDataPage(ParquetColumnChunk chunk)
{
	if (chunk->pageNumber >= chunk->maxPageCount)
	{
		chunk->pages = repalloc(chunk->pages,
				2 * chunk->maxPageCount * sizeof(struct ParquetDataPage_S));
		/* make sure all allocated page memory are zero-filled */
		memset(chunk->pages + chunk->maxPageCount, 0, chunk->maxPageCount * sizeof(struct ParquetDataPage_S));
		chunk->maxPageCount *= 2;
	}

	chunk->currentPage = &chunk->pages[chunk->pageNumber];
	chunk->currentPage->data = NULL;
	chunk->currentPage->finalized = false;
	chunk->currentPage->header = (ParquetPageHeader) palloc0(sizeof(PageMetadata_4C));
	chunk->currentPage->header->page_type = DATA_PAGE;
	chunk->currentPage->header->definition_level_encoding = RLE;
	chunk->currentPage->header->repetition_level_encoding = RLE;
	chunk->currentPage->parquetFile = chunk->parquetFile;

	if (chunk->columnChunkMetadata->d != 0)
	{
		chunk->currentPage->definition_level = palloc0(sizeof(RLEEncoder));

		RLEEncoder_Init(chunk->currentPage->definition_level,
						widthFromMaxInt(chunk->columnChunkMetadata->d));
	}
	if (chunk->columnChunkMetadata->r != 0)
	{
		chunk->currentPage->repetition_level = palloc0(sizeof(RLEEncoder));

		RLEEncoder_Init(chunk->currentPage->repetition_level,
						widthFromMaxInt(chunk->columnChunkMetadata->r));
	}
	/* use BIT_PACK encoding for bool column */
	if (chunk->columnChunkMetadata->type == BOOLEAN)
	{
		chunk->currentPage->bool_values = palloc0(sizeof(ByteBasedBitPackingEncoder));
		BitPack_InitEncoder(chunk->currentPage->bool_values, /*bitWidth=*/1);
	}
	else
	{
		int max_buffer_size = chunk->pageSizeLimit;
		int min_buffer_size = 512;
		if (chunk->estimateChunkSizeRemained > max_buffer_size)
		{
			chunk->currentPage->values_buffer_capacity = max_buffer_size;
			chunk->estimateChunkSizeRemained -= max_buffer_size;
		}
		else if (chunk->estimateChunkSizeRemained < min_buffer_size)
		{
			chunk->currentPage->values_buffer_capacity = min_buffer_size;
		}
		else
		{
			chunk->currentPage->values_buffer_capacity = chunk->estimateChunkSizeRemained;
		}
		chunk->currentPage->values_buffer = palloc0(chunk->currentPage->values_buffer_capacity);
	}

	chunk->pageNumber++;
}

int
appendParquetColumnNull(ParquetColumnChunk columnChunk)
{
	int bytes_added = 0;

	/*if page is null, initialize a new page*/
	if ((columnChunk->pageNumber == 0) || (columnChunk->currentPage == NULL)) {
		addDataPage(columnChunk);
	}

	/* If page size exceeds limit, finalize current data page and add a new one*/
	if (approximatePageSize(columnChunk->currentPage) >= columnChunk->pageSizeLimit)
	{
		bytes_added += finalizeCurrentAndNewPage(columnChunk);
	}

	Assert(columnChunk->currentPage->definition_level != NULL);
	RLEEncoder_WriteInt(columnChunk->currentPage->definition_level, 0);

	if (columnChunk->currentPage->repetition_level != NULL)
	{
		RLEEncoder_WriteInt(columnChunk->currentPage->repetition_level, 0);
	}

	columnChunk->currentPage->header->num_values++;
	columnChunk->columnChunkMetadata->valueCount++;
	return bytes_added;
}

/**
 * Finalize current data page, and then add a new page
 * @columnChunk:		The column chunk which needs to add data
 * @bytes_added:		The number of bytes added
 *
 * return uncompressed bytes added to current row group
 */
int
finalizeCurrentAndNewPage(ParquetColumnChunk columnChunk)
{
	int bytes_added = encodeCurrentPage(columnChunk);

	/*add a new page*/
	addDataPage(columnChunk);

	return bytes_added;
}

/**
 * add a value to a column. includes: adding r to repetition level; adding d to definition level;
 * adding the value itself to page data section
 * @chunk:			the column writer which needs to write data
 * @value:			the value needed to be inserted
 * @r:				the repetition level for the value
 * @d:				the definition level for the value
 *
 * return uncompressed bytes added to current row group
 */
int
appendParquetColumnValue(ParquetColumnChunk chunk,
						 Datum value,
						 int r,
						 int d)
{
	int bytes_added = 0;
	int encoded_len = 0;

	/*if page is null, initialize a new page*/
	if ((chunk->pageNumber == 0) || (chunk->currentPage == NULL))
	{
		addDataPage(chunk);
	}

	encoded_len = encodePlain(value,
							  chunk->currentPage,
							  chunk->columnChunkMetadata->hawqTypeId,
							  chunk->pageSizeLimit);

	if (encoded_len == ENCODE_INVALID_VALUE)
	{
		ereport(ERROR,
				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
				 errmsg("value for column \"%s\" exceeds pagesize %d!",
						chunk->columnChunkMetadata->colName, chunk->pageSizeLimit)));
	}

	if (encoded_len == ENCODE_OUTOF_PAGE)
	{
		bytes_added += finalizeCurrentAndNewPage(chunk);
		encoded_len = encodePlain(value,
								  chunk->currentPage,
								  chunk->columnChunkMetadata->hawqTypeId,
								  chunk->pageSizeLimit);

	}

	bytes_added += encoded_len;

	if (chunk->currentPage->repetition_level != NULL)
	{
		RLEEncoder_WriteInt(chunk->currentPage->repetition_level, r);
	}
	if (chunk->currentPage->definition_level != NULL)
	{
		RLEEncoder_WriteInt(chunk->currentPage->definition_level, d);
	}

	chunk->currentPage->header->num_values++;
	chunk->currentPage->header->uncompressed_page_size += encoded_len;

	chunk->columnChunkMetadata->valueCount++;

	return bytes_added;
}

/*
 * Append null for field.
 *
 * We don't consider UDT currently, therefore we don't have intermediate
 * null value. If one table's attribute is null, all its corresponding
 * columns are null, having (r,d) == (0,0).
 *
 * Return uncompressed bytes added to current row group.
 */
int
appendNullForFields(struct FileField_4C *field,
					ParquetColumnChunk columnChunks,
					int *colIndex)
{
	int bytes_added = 0;

	if (field->num_children == 0)
	{
		bytes_added += appendParquetColumnNull(&columnChunks[*colIndex]);
		*colIndex = *colIndex + 1;
	}
	else
	{
		for (int i = 0; i < field->num_children; i++)
		{
			bytes_added += appendNullForFields(&field->children[i], columnChunks, colIndex);
		}
	}

	return bytes_added;
}

/**
 * Append the value of a hawq field to parquet columns chunks. Should consider embedded types
 * repetition level and definition level calculation.
 * @field:			the hawq field for the value
 * @columnChunk:	the parquet column chunks needed to be inserted into
 * @colIndex:		current parquet column chunk index
 * @value:			the value needed to be inserted
 */
int
appendValueForFields(struct FileField_4C *field,
					 ParquetColumnChunk columnChunks,
					 int *colIndex,
					 Datum value)
{
	int bytes_added = 0;

	/*primitive type, r = 0, d = field->definition_level*/
	if (field->num_children == 0)
	{
		bytes_added += appendParquetColumnValue(columnChunks + (*colIndex),
												value,
												0, field->d);
		*colIndex = *colIndex + 1;
	}
	else
	{
		switch (field->hawqTypeId)
		{
		/* HAWQ built-in embeded type */
		case HAWQ_TYPE_POINT:
			bytes_added += appendParquetColumn_Point(columnChunks,
													 colIndex,
													 DatumGetPointP(value),
													 0, field->d);
			break;
		case HAWQ_TYPE_LSEG:
			bytes_added += appendParquetColumn_Lseg(columnChunks,
													colIndex,
													DatumGetLsegP(value),
													0, field->d);
			break;
		case HAWQ_TYPE_PATH:
		{
			PATH *path = DatumGetPathP(value);
			bytes_added += appendParquetColumn_Path(columnChunks, colIndex,
					path, 0, field->d);
			if(VARATT_IS_EXTENDED((struct varlena *) DatumGetPointer(value)))
				pfree(path);
			break;
		}
		case HAWQ_TYPE_BOX:
			bytes_added += appendParquetColumn_Box(columnChunks,
												   colIndex,
												   DatumGetBoxP(value),
												   0, field->d);
			break;
		case HAWQ_TYPE_POLYGON:
		{
			POLYGON *polygon = DatumGetPolygonP(value);
			bytes_added += appendParquetColumn_Polygon(columnChunks,
													   colIndex,
													   polygon,
													   0, field->d);
			if(VARATT_IS_EXTENDED((struct varlena *) DatumGetPointer(value)))
				pfree(polygon);
			break;
		}
		case HAWQ_TYPE_CIRCLE:
			bytes_added += appendParquetColumn_Circle(columnChunks,
													  colIndex,
													  DatumGetCircleP(value),
													  0, field->d);
			break;

		default:
			/* TODO array type */
			/* TODO UDT */
			Insist(false);
			break;

		}
	}

	return bytes_added;
}

int
appendParquetColumn_Point(ParquetColumnChunk columnChunks, int *colIndex,
						  Point *point, int r, int d)
{
	int bytes_added = 0;

	/* x and y are required, there both r and d remains unchanged. */
	bytes_added += appendParquetColumnValue(columnChunks + (*colIndex),
											Float8GetDatum(point->x), r, d);
	*colIndex = *colIndex + 1;

	bytes_added += appendParquetColumnValue(columnChunks + (*colIndex),
											Float8GetDatum(point->y), r, d);
	*colIndex = *colIndex + 1;

	return bytes_added;
}

int
appendParquetColumn_Lseg(ParquetColumnChunk columnChunks, int *colIndex,
						 LSEG *lseg, int r, int d)
{
	int bytes_added = 0;

	bytes_added += appendParquetColumn_Point(columnChunks, colIndex, lseg->p, r, d);
	bytes_added += appendParquetColumn_Point(columnChunks, colIndex, lseg->p + 1, r, d);

	return bytes_added;
}

int
appendParquetColumn_Path(ParquetColumnChunk columnChunks, int *colIndex,
						 PATH *path, int r, int d)
{
	int i;
	int bytes_added = 0;
	bool is_open = !path->closed;

	/* append is_open column */
	bytes_added += appendParquetColumnValue(columnChunks + (*colIndex),
											BoolGetDatum(is_open), r, d);
	*colIndex += 1;

	/* append points.x column */
	for (i = 0; i < path->npts; ++i)
	{
		if (i == 0)
			bytes_added += appendParquetColumnValue(columnChunks + (*colIndex),
													Float8GetDatum(path->p[i].x),
													0, d + 1);
		else
			bytes_added += appendParquetColumnValue(columnChunks + (*colIndex),
													Float8GetDatum(path->p[i].x),
													r + 1, d + 1);
	}
	*colIndex += 1;

	/* append points.y column */
	for (i = 0; i < path->npts; ++i)
	{
		if (i == 0)
			bytes_added += appendParquetColumnValue(columnChunks + (*colIndex),
													Float8GetDatum(path->p[i].y),
													0, d + 1);
		else
			bytes_added += appendParquetColumnValue(columnChunks + (*colIndex),
													Float8GetDatum(path->p[i].y),
													r + 1, d + 1);
	}
	*colIndex += 1;

	return bytes_added;
}

int
appendParquetColumn_Box(ParquetColumnChunk columnChunks, int *colIndex,
						BOX *box, int r, int d)
{
	int bytes_added = 0;

	bytes_added += appendParquetColumn_Point(columnChunks, colIndex, &box->high, r, d);
	bytes_added += appendParquetColumn_Point(columnChunks, colIndex, &box->low, r, d);

	return bytes_added;
}

int
appendParquetColumn_Polygon(ParquetColumnChunk columnChunks, int *colIndex,
							POLYGON *polygon, int r, int d)
{
	int i;
	int bytes_added = 0;

	/* append boundbox:{x1,y1,x2,y2} columns */
	bytes_added += appendParquetColumn_Box(columnChunks, colIndex, &polygon->boundbox, r, d);

	/* append points:x column */
	for (i = 0; i < polygon->npts; ++i)
	{
		if (i == 0)
			bytes_added += appendParquetColumnValue(columnChunks + (*colIndex),
													Float8GetDatum(polygon->p[i].x),
													0, d + 1);
		else
			bytes_added += appendParquetColumnValue(columnChunks + (*colIndex),
													Float8GetDatum(polygon->p[i].x),
													r + 1, d + 1);
	}
	*colIndex += 1;

	/* append points:y column */
	for (i = 0; i < polygon->npts; ++i)
	{
		if (i == 0)
			bytes_added += appendParquetColumnValue(columnChunks + (*colIndex),
													Float8GetDatum(polygon->p[i].y),
													0, d + 1);
		else
			bytes_added += appendParquetColumnValue(columnChunks + (*colIndex),
													Float8GetDatum(polygon->p[i].y),
													r + 1, d + 1);
	}
	*colIndex += 1;

	return bytes_added;
}

int
appendParquetColumn_Circle(ParquetColumnChunk columnChunks, int *colIndex,
						   CIRCLE *circle,
						   int r, int d)
{
	int bytes_added = 0;

	bytes_added += appendParquetColumn_Point(columnChunks, colIndex, &circle->center, r, d);
	bytes_added += appendParquetColumnValue(columnChunks + (*colIndex), Float8GetDatum(circle->radius), r, d);
	*colIndex += 1;

	return bytes_added;
}

void
appendRowValue(ParquetRowGroup rowgroup,
			   ParquetMetadata parquetmd,
			   Datum* values, bool* nulls)
{
	int bytes_added = 0;

	/*
	 * Append row value column by column.
	 *
	 * One table's column may corresponds to multiple parquet columns
	 * due to nested data type like point, array, UDF, etc.
	 *
	 */
	int colIndex = 0;
	for (int i = 0; i < parquetmd->fieldCount; i++)
	{
		/* for null value, we insert definition level to underlying columns */
		if (nulls[i])
		{
			bytes_added += appendNullForFields(&(parquetmd->pfield[i]),
											   rowgroup->columnChunks,
											   &colIndex);
		}
		/* otherwise the actual value is written, possibly along with r/d */
		else
		{
			bytes_added += appendValueForFields(&parquetmd->pfield[i],
												rowgroup->columnChunks,
												&colIndex,
												values[i]);
		}
	}
	Assert(colIndex == parquetmd->colCount);

	rowgroup->rowGroupMetadata->totalByteSize += bytes_added;
	rowgroup->rowGroupMetadata->rowCount++;
}

/*
 * getTypeName
 *		get name of a type
 *
 * Note: any associated array type is *not* renamed; caller must make
 * another call to handle that case.  Currently this is only used for
 * renaming types associated with tables, for which there are no arrays.
 */
char *
getTypeName(Oid typeOid) {
	Relation pg_type_desc;
	HeapTuple tuple;
	Form_pg_type form;
	cqContext *pcqCtx;
	cqContext cqc;
	char *typeName;
	pg_type_desc = heap_open(TypeRelationId, RowExclusiveLock);

	pcqCtx = caql_addrel(cqclr(&cqc), pg_type_desc);

	tuple = caql_getfirst(
			pcqCtx,
			cql("SELECT typname FROM pg_type "
					" WHERE oid = :1 ",
					ObjectIdGetDatum(typeOid)));

	if (!HeapTupleIsValid(tuple))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT), errmsg("type with OID \"%d\" does not exist", typeOid)));

	form = (Form_pg_type) GETSTRUCT(tuple);
	typeName = (char*) palloc0(strlen(form->typname.data) + 1);
	memcpy(typeName, form->typname.data, strlen(form->typname.data));

	heap_freetuple(tuple);
	heap_close(pg_type_desc, RowExclusiveLock);
	return typeName;
}

/*
 * Before finalize a page, we cannot know the exact number of
 * uncompressed size of a page, due to rle/bitpack encoder buffers
 * some input value.
 *
 * This producure returns an approximate uncompressed page size which
 * mey be a little smaller than the actual size.
 */
int
approximatePageSize(ParquetDataPage page)
{
	int size = page->header->uncompressed_page_size;

	if (page->repetition_level != NULL)
		size += RLEEncoder_Size(page->repetition_level);

	if (page->definition_level != NULL)
		size += RLEEncoder_Size(page->definition_level);

	return size;
}


text:
package com.android.layoutlib.bridge.android;

import com.android.internal.view.IInputContext;
import com.android.internal.view.IInputMethodClient;
import com.android.internal.view.IInputMethodManager;
import com.android.internal.view.InputBindResult;

import android.os.IBinder;
import android.os.RemoteException;
import android.os.ResultReceiver;
import android.text.style.SuggestionSpan;
import android.view.inputmethod.EditorInfo;
import android.view.inputmethod.InputMethodInfo;
import android.view.inputmethod.InputMethodSubtype;

import java.util.List;

/**
 * Basic implementation of IInputMethodManager that does nothing.
 *
 */
public class BridgeIInputMethodManager implements IInputMethodManager {

    @Override
    public void addClient(IInputMethodClient arg0, IInputContext arg1, int arg2, int arg3)
            throws RemoteException {
        // TODO Auto-generated method stub

    }

    @Override
    public void finishInput(IInputMethodClient arg0) throws RemoteException {
        // TODO Auto-generated method stub

    }

    @Override
    public InputMethodSubtype getCurrentInputMethodSubtype() throws RemoteException {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public List<InputMethodInfo> getEnabledInputMethodList() throws RemoteException {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public List<InputMethodSubtype> getEnabledInputMethodSubtypeList(String arg0,
            boolean arg1) throws RemoteException {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public List<InputMethodInfo> getInputMethodList() throws RemoteException {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public InputMethodSubtype getLastInputMethodSubtype() throws RemoteException {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public List getShortcutInputMethodsAndSubtypes() throws RemoteException {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public void hideMySoftInput(IBinder arg0, int arg1) throws RemoteException {
        // TODO Auto-generated method stub

    }

    @Override
    public boolean hideSoftInput(IInputMethodClient arg0, int arg1, ResultReceiver arg2)
            throws RemoteException {
        // TODO Auto-generated method stub
        return false;
    }

    @Override
    public boolean notifySuggestionPicked(SuggestionSpan arg0, String arg1, int arg2)
            throws RemoteException {
        // TODO Auto-generated method stub
        return false;
    }

    @Override
    public void registerSuggestionSpansForNotification(SuggestionSpan[] arg0)
            throws RemoteException {
        // TODO Auto-generated method stub

    }

    @Override
    public void removeClient(IInputMethodClient arg0) throws RemoteException {
        // TODO Auto-generated method stub

    }

    @Override
    public void setAdditionalInputMethodSubtypes(String arg0, InputMethodSubtype[] arg1)
            throws RemoteException {
        // TODO Auto-generated method stub
    }

    @Override
    public boolean setCurrentInputMethodSubtype(InputMethodSubtype arg0) throws RemoteException {
        // TODO Auto-generated method stub
        return false;
    }

    @Override
    public void setImeWindowStatus(IBinder arg0, int arg1, int arg2) throws RemoteException {
        // TODO Auto-generated method stub

    }

    @Override
    public void setInputMethod(IBinder arg0, String arg1) throws RemoteException {
        // TODO Auto-generated method stub

    }

    @Override
    public void setInputMethodAndSubtype(IBinder arg0, String arg1, InputMethodSubtype arg2)
            throws RemoteException {
        // TODO Auto-generated method stub

    }

    @Override
    public boolean setInputMethodEnabled(String arg0, boolean arg1) throws RemoteException {
        // TODO Auto-generated method stub
        return false;
    }

    @Override
    public void showInputMethodAndSubtypeEnablerFromClient(IInputMethodClient arg0, String arg1)
            throws RemoteException {
        // TODO Auto-generated method stub

    }

    @Override
    public void showInputMethodPickerFromClient(IInputMethodClient arg0) throws RemoteException {
        // TODO Auto-generated method stub

    }

    @Override
    public void showMySoftInput(IBinder arg0, int arg1) throws RemoteException {
        // TODO Auto-generated method stub

    }

    @Override
    public boolean showSoftInput(IInputMethodClient arg0, int arg1, ResultReceiver arg2)
            throws RemoteException {
        // TODO Auto-generated method stub
        return false;
    }

    @Override
    public InputBindResult startInput(IInputMethodClient client, IInputContext inputContext,
            EditorInfo attribute, int controlFlags) throws RemoteException {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public boolean switchToLastInputMethod(IBinder arg0) throws RemoteException {
        // TODO Auto-generated method stub
        return false;
    }

    @Override
        public boolean switchToNextInputMethod(IBinder arg0, boolean arg1) throws RemoteException {
        // TODO Auto-generated method stub
        return false;
    }

    @Override
        public boolean shouldOfferSwitchingToNextInputMethod(IBinder arg0) throws RemoteException {
        // TODO Auto-generated method stub
        return false;
    }

    @Override
    public void updateStatusIcon(IBinder arg0, String arg1, int arg2) throws RemoteException {
        // TODO Auto-generated method stub

    }

    @Override
    public InputBindResult windowGainedFocus(IInputMethodClient client, IBinder windowToken,
            int controlFlags, int softInputMode, int windowFlags, EditorInfo attribute,
            IInputContext inputContext) throws RemoteException {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public IBinder asBinder() {
        // TODO Auto-generated method stub
        return null;
    }
}

redpajama_set_name: RedPajamaGithub


text:
<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">

<title>thobe.logfileviewer.kernel.source.err Class Hierarchy</title>
<meta name="date" content="2015-03-22">
<link rel="stylesheet" type="text/css" href="../../../../../stylesheet.css" title="Style">

JavaScript is disabled on your browser.

Overview

Package

Class

Use

Tree

Deprecated

Index

Help

Prev

Next

Frames

No Frames

All Classes

Hierarchy For Package thobe.logfileviewer.kernel.source.err

Package Hierarchies:

All Packages

Class Hierarchy

java.lang.Object

java.lang.Throwable (implements java.io.Serializable)

java.lang.Exception

thobe.logfileviewer.kernel.source.err.LogLineBufferException

thobe.logfileviewer.kernel.source.err.LogStreamException

thobe.logfileviewer.kernel.source.err.LogStreamTimeoutException

Overview

Package

Class

Use

Tree

Deprecated

Index

Help

Prev

Next

Frames

No Frames

All Classes


text:
package vmutils

import (
	"encoding/binary"
)

type BytesReader struct {
	byteOrder binary.ByteOrder
	data      []byte
	position  int
}

func NewBytesReader(data []byte, byteOrder binary.ByteOrder) BytesReader {
	return BytesReader{
		byteOrder: byteOrder,
		data:      data,
		position:  0,
	}
}

func (reader *BytesReader) Position() int {
	return reader.position
}

func (reader *BytesReader) ReadUint8() uint8 {
	i := reader.data[reader.position]
	reader.position++
	return i
}

func (reader *BytesReader) ReadUint16() uint16 {
	i := reader.byteOrder.Uint16(reader.data[reader.position:])
	reader.position += 2
	return i
}

func (reader *BytesReader) ReadUint32() uint32 {
	i := reader.byteOrder.Uint32(reader.data[reader.position:])
	reader.position += 4
	return i
}

func (reader *BytesReader) ReadUint64() uint64 {
	i := reader.byteOrder.Uint64(reader.data[reader.position:])
	reader.position += 8
	return i
}

func (reader *BytesReader) ReadBytes(n int) []byte {
	bytes := reader.data[reader.position : reader.position+n]
	reader.position += n
	return bytes
}


text:
BEGIN;
set client_min_messages to warning;
set local search_path = 'bi_temp_tables','bitemporal_internal','public';
set local TimeZone  = 'UTC';

SELECT plan(27);

select  unialike( current_setting('search_path'), '%temporal_relationships%'
  ,'temporal_relationships should NOT be on search_path for these tests' );

select lives_ok($$
    create schema bi_temp_tables
$$, 'create schema');

select lives_ok($$
  create table bi_temp_tables.devices_manual (
      device_id_key serial NOT NULL
    , device_id integer NOT NULL
    , effective tstzrange
    , asserted tstzrange
    , device_descr text
    , row_created_at timestamptz NOT NULL DEFAULT now()
    , CONSTRAINT devices_device_id_asserted_effective_excl EXCLUDE
      USING gist (device_id WITH =, asserted WITH &&, effective WITH &&)
  )
$$, 'create devices manual');

select lives_ok($$select * from bitemporal_internal.ll_create_bitemporal_table('bi_temp_tables','devices',
'device_id integer, device_descr text', 'device_id')
$$, 'create devices');

select lives_ok($$
insert into  bi_temp_tables.devices( device_id , effective, asserted, device_descr )
values
 (1, '[01-01-2015, infinity)', '[01-01-2015, infinity)','descr2')
,(5, '[2015-12-01 00:00:00-06,2015-12-31 00:00:00+00)', '[2015-12-01 00:00:00-06,2015-12-31 00:00:00+00)', 'test_5')
$$, 'insert data into devices');

---non-temp for test

select lives_ok($$
  create table bi_temp_tables.devices_non_temp(
    device_id_key serial,
    device_id int4
  );
  create table bi_temp_tables.devices_temp(
   device_id_key serial,
   device_id int4,
   effective tsrange,
   EXCLUDE USING gist (device_id WITH =,  effective WITH &&)
  )
$$, 'double create');

select is( ll_is_bitemporal_table('bi_temp_tables.devices_temp'), false
      , 'is bitemporal table? devices_temp');

select is( ll_is_bitemporal_table('bi_temp_tables.devices')
    , true  , 'is bitemporal table? devices');

select is( ll_is_bitemporal_table('bi_temp_tables.devices_manual')
    , true  , 'is bitemporal table? devices_manual');

select is( ll_is_bitemporal_table('bi_temp_tables.devices_non_temp')
    , false , 'is bitemporal table? devices_non_temp');

select is( ll_is_bitemporal_table('bi_temp_tables.dev')
    , false , 'is bitemporal table? dev');

----test insert:
 select results_eq ($q$
  select bitemporal_internal.ll_bitemporal_insert('bi_temp_tables.devices',
  'device_id , device_descr', $$'11', 'new_descr'$$, '[01-01-2016, infinity)', '[01-02-2016, infinity)' )
$q$,
$v$ values(1) $v$
,'bitemporal insert'
);

-- select * from bi_temp_tables.devices ;

select results_eq($q$ select device_id, device_descr, effective, asserted
from bi_temp_tables.devices where device_id =11 $q$
, $v$
values
( 11
,'new_descr'::text
  ,'["2016-01-01 00:00:00+00",infinity)'::temporal_relationships.timeperiod
  ,'["2016-01-02 00:00:00+00",infinity)'::temporal_relationships.timeperiod
)
$v$
,'bitemporal insert, returns select '
);

---test correction

select results_eq($q$select * from bitemporal_internal.ll_bitemporal_correction('bi_temp_tables','devices',
'device_descr',
$$'updated_descr_11'$$,
'device_id' ,
'11',
'[01-01-2016, infinity)' ::temporal_relationships.timeperiod)$q$,
$v$ values(1) $v$
,'bitemporal correction'
);

select results_eq($q$
  select device_descr
        from bi_temp_tables.devices where device_id = 11 and upper(asserted)='infinity' and effective='[01-01-2016, infinity)'
$q$
, $v$
values
('updated_descr_11'::text)
$v$
,'select after bitemporal correction - old'
);

select results_eq($q$
  select device_descr
        from bi_temp_tables.devices where device_id = 11 and lower(asserted)='2016-01-02'
         and effective='[01-01-2016, infinity)'
$q$
, $v$
values
('new_descr'::text)
$v$
,'select after bitemporal correction - new'
);

select results_eq($q$
    select * from
  bitemporal_internal.ll_bitemporal_list_of_fields('bi_temp_tables.devices')
$q$
, $v$
values
(
ARRAY['device_id','device_descr']
)
$v$
,'list of fields'
);

---correct test:

select results_eq($q$
select *  from bitemporal_internal.ll_bitemporal_update('bi_temp_tables','devices'
,'device_descr'
,$$'descr starting from jan 1'$$
,'device_id'
,$$1$$
,'[2018-01-01, infinity)'
, '[3016-01-01, infinity)') $q$,
$v$ values(1) $v$
,'bitemporal update - correct'
);

select results_eq ($q$
select bitemporal_internal.ll_bitemporal_insert('bi_temp_tables.devices',
  'device_id , device_descr', $$'10', 'descr_crean_insert'$$, '[01-01-2017, infinity)', '[3017-07-09 21:59:58.993815-05, infinity)' )
$q$,
$v$ values(1) $v$
,'bitemporal insert for correction'
);

select results_eq($q$select * from bitemporal_internal.ll_bitemporal_correction('bi_temp_tables','devices',
'device_descr',
$$'descr_10_corr_on_place'$$,
'device_id' ,
'10',
'[01-01-2017, infinity)'::temporal_relationships.timeperiod , '3017-07-09 21:59:58.993815-05'
)$q$,
$v$ values(1) $v$
,'bitemporal correction on place'
);

select results_eq($q$
  select device_descr
        from bi_temp_tables.devices where device_id = 10
         and effective='[01-01-2017, infinity)'
$q$
, $v$
values
('descr_10_corr_on_place'::text)
$v$
,'select after bitemporal correction on place'
);

select results_eq($q$select * from bitemporal_internal.ll_bitemporal_correction('bi_temp_tables','devices',
'device_descr',
$$'descr_10_corr_with_new-record'$$,
'device_id' ,
'10',
'[01-01-2017, infinity)' ::temporal_relationships.timeperiod, '3017-07-09 22:10:58.993815-05'
 )$q$,
$v$ values(1) $v$
,'bitemporal correction new record'
);

select results_eq($q$
  select device_descr
        from bi_temp_tables.devices where device_id = 10
         and effective='[01-01-2017, infinity)' and upper(asserted)='infinity'
$q$
, $v$
values
('descr_10_corr_with_new-record'::text)
$v$
,'select after bitemporal correction new record'
);
---test update with error:
/*

select * from bitemporal_internal.ll_bitemporal_update('bi_temp_tables.devices'
,'device_descr'
,$$'descr starting from jan 1'$$
,'device_id'
,$$8$$
,'[2016-01-01, infinity)'
, '[2016-01-02, infinity)')
;

---output:

ERROR:  Asserted interval starts in the past or has a finite end: ["2016-01-02 00:00:00-06",infinity)

Exactly the same test should be performed for inactivate
*/

/* should probably include the test of bitemporal_internal.ll_check_bitemporal_update_conditions - if it returns zero,
then

ERROR:  Nothing to update, use INSERT or check effective: ["2015-01-02 00:00:00-06","2015-02-02")

Exactly the same test should be performed for inactivate

*/

select results_eq($q$select * from bitemporal_internal.ll_bitemporal_inactivate('bi_temp_tables','devices'
,'device_id'
,$$11$$
,'[3016-02-02, infinity)'
, '[3016-03-01, infinity)')  $q$,
$v$ values(1) $v$
,'bitemporal inactivate - correct'
);

select results_eq($q$select count(*)::integer from bi_temp_tables.devices
where device_id=11
and  '[3016-03-16,  3016-03-16]'<@ asserted
and '[3016-02-04, 3016-02-04]' <@ effective $q$,
$v$ values(0::integer) $v$,'bitemporal inactivate no active rows');

---delete:

select results_eq($q$select * from bitemporal_internal.ll_bitemporal_delete('bi_temp_tables.devices'
,'device_id'
,$$1$$
, '[3016-04-04 21:30, infinity)')  $q$,
$v$ values(2) $v$
,'bitemporal delete');

select results_eq($q$select count(*)::integer from bi_temp_tables.devices
where device_id=1
and '[3016-04-05, 3016-04-05]' <@ asserted $q$,
$v$ values(0::integer) $v$
,'bitemporal delete no active rows');

SELECT * FROM finish();
ROLLBACK;
-- vim: set filetype=pgsql expandtab tabstop=2 shiftwidth=2:


text:
using System;
using System.Linq;
using System.Reflection;
using Cuemon.Extensions.IO;
using Cuemon.Extensions.Xunit;
using Cuemon.Reflection;
using Xunit;
using Xunit.Abstractions;

namespace Cuemon
{
    public class StringReplacePairTest : Test
    {
        private static readonly string LoremIpsum = Decorator.Enclose(typeof(StringReplacePairTest).Assembly).GetManifestResources("LoremIpsum.txt", ManifestResourceMatch.ContainsName).Single().Value.ToEncodedString();

        public StringReplacePairTest(ITestOutputHelper output) : base(output)
        {
        }

        [Fact]
        public void ReplaceAll_ShouldReplaceAllOccurrencesOfOldValueWithNewValue_UsingOrdinalComparison()
        {
            var comparison = StringComparison.Ordinal;
            var loremIpsumWords = LoremIpsum.Split(' ');
            var cuemonCountBeforeSut = loremIpsumWords.Count(s => s.Contains("cuemon", comparison));
            var utCountBeforeSut = loremIpsumWords.Count(s => s.Contains("ut", comparison));
            var sut = StringReplacePair.ReplaceAll(LoremIpsum, "ut", "cuemon", comparison);
            var sutWords = sut.Split(' ');
            var utCountAfterSut = sutWords.Count(s => s.Contains("ut", comparison));
            var cuemonCountAfterSut = sutWords.Count(s => s.Contains("cuemon", comparison));
            var loremIpsumDifference = sutWords.Except(loremIpsumWords).ToList();
            var sutDifference = loremIpsumWords.Except(sutWords).ToList();

            Assert.NotEqual(loremIpsumWords, sutWords);

            Assert.Equal(loremIpsumDifference.Select(s => s.Replace("cuemon", "", comparison)), sutDifference.Select(s => s.Replace("ut", "", comparison)));
            Assert.Equal(loremIpsumDifference.Count, sutDifference.Count);
            Assert.Equal(loremIpsumWords.Length, sutWords.Length);
            Assert.Equal(utCountBeforeSut, cuemonCountAfterSut);
            Assert.Equal(cuemonCountBeforeSut, utCountAfterSut);

            TestOutput.WriteLine(sut);
        }

        [Fact]
        public void ReplaceAll_ShouldReplaceAllOccurrencesOfOldValueWithNewValue()
        {
            var comparison = StringComparison.OrdinalIgnoreCase;
            var loremIpsumWords = LoremIpsum.Split(' ');
            var cuemonCountBeforeSut = loremIpsumWords.Count(s => s.Contains("cuemon", comparison));
            var utCountBeforeSut = loremIpsumWords.Count(s => s.Contains("ut", comparison));
            var sut = StringReplacePair.ReplaceAll(LoremIpsum, "ut", "cuemon"); // default is ordinal-ignore-case
            var sutWords = sut.Split(' ');
            var utCountAfterSut = sutWords.Count(s => s.Contains("ut", comparison));
            var cuemonCountAfterSut = sutWords.Count(s => s.Contains("cuemon", comparison));
            var loremIpsumDifference = sutWords.Except(loremIpsumWords).ToList();
            var sutDifference = loremIpsumWords.Except(sutWords).ToList();

            Assert.NotEqual(loremIpsumWords, sutWords);
            Assert.NotEqual(loremIpsumDifference.Select(s => s.Replace("cuemon", "", comparison)), sutDifference.Select(s => s.Replace("ut", "", comparison)));
            Assert.NotEqual(loremIpsumDifference.Count, sutDifference.Count);

            Assert.Equal(loremIpsumWords.Length, sutWords.Length);
            Assert.Equal(utCountBeforeSut, cuemonCountAfterSut);
            Assert.Equal(cuemonCountBeforeSut, utCountAfterSut);

            TestOutput.WriteLine(sut);
        }

        [Fact]
        public void RemoveAll_ShouldRemoveAllOccurrencesOfFragments_UsingOrdinalIgnoreCaseComparison()
        {
            var comparison = StringComparison.OrdinalIgnoreCase;
            var loremIpsumWords = LoremIpsum.Split(' ');
            var ametCountBeforeSut = loremIpsumWords.Count(s => s.Contains("amet", comparison));
            var scelerisqueCountBeforeSut = loremIpsumWords.Count(s => s.Contains("scelerisque", comparison));
            var sut = StringReplacePair.RemoveAll(LoremIpsum, comparison, "amet", "scelerisque");
            var sutWords = sut.Split(' ');
            var ametCountAfterSut = sutWords.Count(s => s.Contains("amet", comparison));
            var scelerisqueCountAfterSut = sutWords.Count(s => s.Contains("scelerisque", comparison));

            Assert.NotEqual(loremIpsumWords, sutWords);

            Assert.Equal(loremIpsumWords.Length, sutWords.Length);

            Assert.Equal(111, ametCountBeforeSut);
            Assert.Equal(0, ametCountAfterSut);
            Assert.Equal(53, scelerisqueCountBeforeSut);
            Assert.Equal(0, scelerisqueCountAfterSut);

            TestOutput.WriteLine(sut);
        }

        [Fact]
        public void RemoveAll_ShouldRemoveAllOccurrencesOfFragments()
        {
            var comparison = StringComparison.Ordinal;
            var loremIpsumWords = LoremIpsum.Split(' ');
            var ametCountBeforeSut = loremIpsumWords.Count(s => s.Contains("amet", comparison));
            var scelerisqueCountBeforeSut = loremIpsumWords.Count(s => s.Contains("scelerisque", comparison));
            var sut = StringReplacePair.RemoveAll(LoremIpsum, "amet", "scelerisque"); // default is ordinal
            var sutWords = sut.Split(' ');
            var ametCountAfterSut = sutWords.Count(s => s.Contains("amet", comparison));
            var scelerisqueCountAfterSut = sutWords.Count(s => s.Contains("scelerisque", comparison));

            Assert.NotEqual(loremIpsumWords, sutWords);

            Assert.Equal(loremIpsumWords.Length, sutWords.Length);

            Assert.Equal(100, ametCountBeforeSut);
            Assert.Equal(0, ametCountAfterSut);
            Assert.Equal(49, scelerisqueCountBeforeSut);
            Assert.Equal(0, scelerisqueCountAfterSut);

            TestOutput.WriteLine(sut);
        }
    }
}


text:
<?php

namespace Yoda\EventBundle\Controller;

use Symfony\Component\HttpFoundation\Response;
use Yoda\EventBundle\Controller\Controller;

class DefaultController extends Controller
{
    public function indexAction($name, $count)
    {
        $em = $this->getDoctrine()->getManager();
        $repo = $em->getRepository('EventBundle:Event');
        $event = $repo->findOneBy(array('name' => 'awesome event'));
        return $this->render('EventBundle:Default:index.html.twig', array(
            'name' => $name,
            'count' => $count,
            'event' => $event
        ));
//        $arr = array(
//            'status' => 'good',
//            'message' => 'Keep smiling'
//        );
//        $response = new Response(json_encode($arr));
//        $response -> headers -> set('Content-Type', 'application/json');
//        return $response;

    }
}


text: export { TableOfContents20 as default } from "../../";


text:
namespace DentistSpace.Web.Infrastructure.Constants
{
    public class CacheConstants
    {
        public const int LongCaching = 60 * 30;
        public const int MediumCaching = 60 * 20;
        public const int ShortCaching = 60 * 10;
    }
}


text:
package org.apache.struts2.result;

import com.opensymphony.xwork2.ActionInvocation;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import javax.servlet.http.HttpServletResponse;
import java.io.InputStream;
import java.io.OutputStream;

/**
 *
 *

 * A custom Result type for sending raw data (via an InputStream) directly to the
 * HttpServletResponse. Very useful for allowing users to download content.
 *

 *
 *

 * This result type takes the following parameters:
 *

 *
 *
 *

 *
 *
contentType - the stream mime-type as sent to the web browser
 * (default = text/plain).

 *
 *
contentLength - the stream length in bytes (the browser displays a
 * progress bar).

 *
 *
contentDisposition - the content disposition header value for
 * specifing the file name (default = inline, values are typically
 * attachment;filename="document.pdf".

 *
 *
inputName - the name of the InputStream property from the chained
 * action (default = inputStream).

 *
 *
bufferSize - the size of the buffer to copy from input to output
 * (default = 1024).

 *
 *
allowCaching if set to 'false' it will set the headers 'Pragma' and 'Cache-Control'
 * to 'no-cahce', and prevent client from caching the content. (default = true)
 *
 *
contentCharSet if set to a string, ';charset=value' will be added to the
 * content-type header, where value is the string set. If set to an expression, the result
 * of evaluating the expression will be used. If not set, then no charset will be set on
 * the header

 *

 *
 *
These parameters can also be set by exposing a similarly named getter method on your Action.  For example, you can
 * provide getContentType() to override that parameter for the current action.

 *
 *
 *

 * Example:
 *

 *
 *

 * <result name="success" type="stream">
 *   <param name="contentType">image/jpeg</param>
 *   <param name="inputName">imageStream</param>
 *   <param name="contentDisposition">attachment;filename="document.pdf"</param>
 *   <param name="bufferSize">1024</param>
 * </result>
 *

 *
 */
public class StreamResult extends StrutsResultSupport {

    private static final long serialVersionUID = -1468409635999059850L;

    protected static final Logger LOG = LogManager.getLogger(StreamResult.class);

    public static final String DEFAULT_PARAM = "inputName";

    protected String contentType = "text/plain";
    protected String contentLength;
    protected String contentDisposition = "inline";
    protected String contentCharSet ;
    protected String inputName = "inputStream";
    protected InputStream inputStream;
    protected int bufferSize = 1024;
    protected boolean allowCaching = true;

    public StreamResult() {
        super();
    }

    public StreamResult(InputStream in) {
        this.inputStream = in;
    }

     /**
     * @return Returns the whether or not the client should be requested to allow caching of the data stream.
     */
    public boolean getAllowCaching() {
        return allowCaching;
    }

    /**
     * Set allowCaching to <tt>false</tt> to indicate that the client should be requested not to cache the data stream.
     * This is set to <tt>false</tt> by default
     *
     * @param allowCaching Enable caching.
     */
    public void setAllowCaching(boolean allowCaching) {
        this.allowCaching = allowCaching;
    }

    /**
     * @return Returns the bufferSize.
     */
    public int getBufferSize() {
        return (bufferSize);
    }

    /**
     * @param bufferSize The bufferSize to set.
     */
    public void setBufferSize(int bufferSize) {
        this.bufferSize = bufferSize;
    }

    /**
     * @return Returns the contentType.
     */
    public String getContentType() {
        return (contentType);
    }

    /**
     * @param contentType The contentType to set.
     */
    public void setContentType(String contentType) {
        this.contentType = contentType;
    }

    /**
     * @return Returns the contentLength.
     */
    public String getContentLength() {
        return contentLength;
    }

    /**
     * @param contentLength The contentLength to set.
     */
    public void setContentLength(String contentLength) {
        this.contentLength = contentLength;
    }

    /**
     * @return Returns the Content-disposition header value.
     */
    public String getContentDisposition() {
        return contentDisposition;
    }

    /**
     * @param contentDisposition the Content-disposition header value to use.
     */
    public void setContentDisposition(String contentDisposition) {
        this.contentDisposition = contentDisposition;
    }

    /**
     * @return Returns the charset specified by the user
     */
    public String getContentCharSet() {
        return contentCharSet;
    }

    /**
     * @param contentCharSet the charset to use on the header when sending the stream
     */
    public void setContentCharSet(String contentCharSet) {
        this.contentCharSet = contentCharSet;
    }

    /**
     * @return Returns the inputName.
     */
    public String getInputName() {
        return (inputName);
    }

    /**
     * @param inputName The inputName to set.
     */
    public void setInputName(String inputName) {
        this.inputName = inputName;
    }

    /**
     * @see StrutsResultSupport#doExecute(java.lang.String, com.opensymphony.xwork2.ActionInvocation)
     */
    protected void doExecute(String finalLocation, ActionInvocation invocation) throws Exception {
        LOG.debug("Find the Response in context");

        OutputStream oOutput = null;

        try {
            if (inputStream == null) {
                LOG.debug("Find the inputstream from the invocation variable stack");
                inputStream = (InputStream) invocation.getStack().findValue(conditionalParse(inputName, invocation));
            }

            if (inputStream == null) {
                String msg = ("Can not find a java.io.InputStream with the name [" + inputName + "] in the invocation stack. " +
                    "Check the <param name=\"inputName\"> tag specified for this action.");
                LOG.error(msg);
                throw new IllegalArgumentException(msg);
            }

            HttpServletResponse oResponse = (HttpServletResponse) invocation.getInvocationContext().get(HTTP_RESPONSE);

            LOG.debug("Set the content type: {};charset{}", contentType, contentCharSet);
            if (contentCharSet != null && ! contentCharSet.equals("")) {
                oResponse.setContentType(conditionalParse(contentType, invocation)+";charset="+conditionalParse(contentCharSet, invocation));
            } else {
                oResponse.setContentType(conditionalParse(contentType, invocation));
            }

            LOG.debug("Set the content length: {}", contentLength);
            if (contentLength != null) {
                String _contentLength = conditionalParse(contentLength, invocation);
                int _contentLengthAsInt;
                try {
                    _contentLengthAsInt = Integer.parseInt(_contentLength);
                    if (_contentLengthAsInt >= 0) {
                        oResponse.setContentLength(_contentLengthAsInt);
                    }
                } catch(NumberFormatException e) {
                    LOG.warn("failed to recognize {} as a number, contentLength header will not be set", _contentLength, e);
                }
            }

            LOG.debug("Set the content-disposition: {}", contentDisposition);
            if (contentDisposition != null) {
                oResponse.addHeader("Content-Disposition", conditionalParse(contentDisposition, invocation));
            }

            LOG.debug("Set the cache control headers if necessary: {}", allowCaching);
            if (!allowCaching) {
                oResponse.addHeader("Pragma", "no-cache");
                oResponse.addHeader("Cache-Control", "no-cache");
            }

            oOutput = oResponse.getOutputStream();

            LOG.debug("Streaming result [{}] type=[{}] length=[{}] content-disposition=[{}] charset=[{}]",
                    inputName, contentType, contentLength, contentDisposition, contentCharSet);

        	LOG.debug("Streaming to output buffer +++ START +++");
            byte[] oBuff = new byte[bufferSize];
            int iSize;
            while (-1 != (iSize = inputStream.read(oBuff))) {
                LOG.debug("Sending stream ... {}", iSize);
                oOutput.write(oBuff, 0, iSize);
            }
        	LOG.debug("Streaming to output buffer +++ END +++");

            // Flush
            oOutput.flush();
        } finally {
            if (inputStream != null) {
                inputStream.close();
            }
            if (oOutput != null) {
                oOutput.close();
            }
        }
    }

}


text:
pushon
======

Anyone who would like to invest in pulling me out of the dependency homeless situation I have been trapped in for so long can pleage your stakes into a successful amd prosperous business future that will enable me to reward those of good intention.


text:
<?php

namespace SebastianBergmann\CodeCoverage\Report\Xml;

class File
{
    /**
     * @var \DOMDocument
     */
    protected $dom;

    /**
     * @var \DOMElement
     */
    protected $contextNode;

    public function __construct(\DOMElement $context)
    {
        $this->dom         = $context->ownerDocument;
        $this->contextNode = $context;
    }

    public function getTotals()
    {
        $totalsContainer = $this->contextNode->firstChild;

        if (!$totalsContainer) {
            $totalsContainer = $this->contextNode->appendChild(
                $this->dom->createElementNS(
                    'http://schema.phpunit.de/coverage/1.0',
                    'totals'
                )
            );
        }

        return new Totals($totalsContainer);
    }

    public function getLineCoverage($line)
    {
        $coverage = $this->contextNode->getElementsByTagNameNS(
            'http://schema.phpunit.de/coverage/1.0',
            'coverage'
        )->item(0);

        if (!$coverage) {
            $coverage = $this->contextNode->appendChild(
                $this->dom->createElementNS(
                    'http://schema.phpunit.de/coverage/1.0',
                    'coverage'
                )
            );
        }

        $lineNode = $coverage->appendChild(
            $this->dom->createElementNS(
                'http://schema.phpunit.de/coverage/1.0',
                'line'
            )
        );

        return new Coverage($lineNode, $line);
    }
}


text:
package org.osgi.enroute.examples.microservice.dao.impl;
public interface PersonTable {

	String TABLE_NAME = "PERSONS";

	String SQL_SELECT_ALL_PERSONS = "SELECT * FROM " + TABLE_NAME;

	String SQL_DELETE_PERSON_BY_PK = "DELETE FROM " + TABLE_NAME + " where PERSON_ID=?";

	String SQL_SELECT_PERSON_BY_PK = "SELECT * FROM " + TABLE_NAME + " where PERSON_ID=?";

	String SQL_INSERT_PERSON = "INSERT INTO " + TABLE_NAME + "(FIRST_NAME,LAST_NAME) VALUES(?,?)";

	String SQL_UPDATE_PERSON_BY_PK = "UPDATE " + TABLE_NAME + " SET FIRST_NAME=?, LAST_NAME=? WHERE PERSON_ID=?";

	String PERSON_ID = "person_id";

	String FIRST_NAME = "first_name";

	String LAST_NAME = "last_name";

	String INIT = "CREATE TABLE IF NOT EXISTS " + TABLE_NAME + " (" //
			+ "person_id bigint GENERATED BY DEFAULT AS IDENTITY," //
			+ "first_name varchar(255) NOT NULL," //
			+ "last_name varchar(255) NOT NULL," //
			+ "PRIMARY KEY (person_id)" //
			+ ") ;";
}